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

Moved bot to kdb-bot #70

Browse Source
This commit is contained in:
Sven Heidemann
2022-10-15 23:17:45 +02:00
parent 029b46d7de
commit ead3f69a69
216 changed files with 0 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
kdb-bot/src/modules/__init__.py Normal file
View File

@@ -0,0 +1 @@
# imports

26
kdb-bot/src/modules/admin/__init__.py Normal file
View File

@@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
"""
bot Keksdose bot
~~~~~~~~~~~~~~~~~~~
Discord bot for the Keksdose discord Server
:copyright: (c) 2022 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'modules.admin'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2022 sh-edraft.de'
__version__ = '0.2.3'
from collections import namedtuple
# imports:
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='0', minor='2', micro='3')

46
kdb-bot/src/modules/admin/admin.json Normal file
View File

@@ -0,0 +1,46 @@
{
"ProjectSettings": {
"Name": "admin",
"Version": {
"Major": "0",
"Minor": "0",
"Micro": "0"
},
"Author": "",
"AuthorEmail": "",
"Description": "",
"LongDescription": "",
"URL": "",
"CopyrightDate": "",
"CopyrightName": "",
"LicenseName": "",
"LicenseDescription": "",
"Dependencies": [
"cpl-core>=2022.10.0.post5"
],
"DevDependencies": [
"cpl-cli==2022.10.0"
],
"PythonVersion": ">=3.10.4",
"PythonPath": {
"linux": ""
},
"Classifiers": []
},
"BuildSettings": {
"ProjectType": "library",
"SourcePath": "",
"OutputPath": "../../dist",
"Main": "admin.main",
"EntryPoint": "admin",
"IncludePackageData": false,
"Included": [],
"Excluded": [
"*/__pycache__",
"*/logs",
"*/tests"
],
"PackageData": {},
"ProjectReferences": []
}
}

24
kdb-bot/src/modules/admin/admin_module.py Normal file
View File

@@ -0,0 +1,24 @@
from cpl_core.configuration import ConfigurationABC
from cpl_core.dependency_injection import ServiceCollectionABC
from cpl_core.environment import ApplicationEnvironmentABC
from cpl_discord.service.discord_collection_abc import DiscordCollectionABC
from bot_core.abc.module_abc import ModuleABC
from bot_core.configuration.feature_flags_enum import FeatureFlagsEnum
from modules.admin.command.restart_command import RestartCommand
from modules.admin.command.shutdown_command import ShutdownCommand
class AdminModule(ModuleABC):
def __init__(self, dc: DiscordCollectionABC):
ModuleABC.__init__(self, dc, FeatureFlagsEnum.admin_module)
def configure_configuration(self, config: ConfigurationABC, env: ApplicationEnvironmentABC):
pass
def configure_services(self, services: ServiceCollectionABC, env: ApplicationEnvironmentABC):
# commands
self._dc.add_command(RestartCommand)
self._dc.add_command(ShutdownCommand)
# events

26
kdb-bot/src/modules/admin/command/__init__.py Normal file
View File

@@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
"""
bot Keksdose bot
~~~~~~~~~~~~~~~~~~~
Discord bot for the Keksdose discord Server
:copyright: (c) 2022 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'modules.admin.command'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2022 sh-edraft.de'
__version__ = '0.2.3'
from collections import namedtuple
# imports:
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='0', minor='2', micro='3')

64
kdb-bot/src/modules/admin/command/restart_command.py Normal file
View File

@@ -0,0 +1,64 @@
import asyncio
import discord
from cpl_core.configuration import ConfigurationABC
from cpl_discord.command import DiscordCommandABC
from cpl_discord.service import DiscordBotServiceABC
from cpl_translation import TranslatePipe
from discord.ext import commands
from discord.ext.commands import Context
from bot_core.abc.client_utils_service_abc import ClientUtilsServiceABC
from bot_core.abc.message_service_abc import MessageServiceABC
from bot_core.configuration.bot_settings import BotSettings
from bot_core.logging.command_logger import CommandLogger
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: ClientUtilsServiceABC,
translate: TranslatePipe,
permissions: PermissionServiceABC,
settings: BotSettings
):
DiscordCommandABC.__init__(self)
self._logger = logger
self._config = config
self._message_service = message_service
self._bot = bot
self._client_utils = client_utils
self._t = translate
self._permissions = permissions
self._settings = settings
self._logger.trace(__name__, f'Loaded command service: {type(self).__name__}')
@commands.hybrid_command()
@commands.guild_only()
async def restart(self, ctx: Context):
self._logger.debug(__name__, f'Received command restart {ctx}')
if not await self._client_utils.check_if_bot_is_ready_yet_and_respond(ctx):
return
self._client_utils.received_command(ctx.guild.id)
if not self._permissions.is_member_moderator(ctx.author):
await self._message_service.send_ctx_msg(ctx, self._t.transform('common.no_permission_message'))
self._logger.trace(__name__, f'Finished restart command')
return
self._config.add_configuration('IS_RESTART', 'true')
await self._client_utils.presence_game('common.presence.restart')
await asyncio.sleep(self._settings.wait_for_restart)
await self._message_service.send_ctx_msg(ctx, self._t.transform('modules.admin.restart_message'))
await self._bot.stop_async()
self._logger.trace(__name__, f'Finished restart command')

63
kdb-bot/src/modules/admin/command/shutdown_command.py Normal file
View File

@@ -0,0 +1,63 @@
import asyncio
import discord
from cpl_core.configuration import ConfigurationABC
from cpl_discord.command import DiscordCommandABC
from cpl_discord.service import DiscordBotServiceABC
from cpl_translation import TranslatePipe
from discord.ext import commands
from discord.ext.commands import Context
from bot_core.abc.client_utils_service_abc import ClientUtilsServiceABC
from bot_core.abc.message_service_abc import MessageServiceABC
from bot_core.configuration.bot_settings import BotSettings
from bot_core.logging.command_logger import CommandLogger
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: ClientUtilsServiceABC,
translate: TranslatePipe,
permissions: PermissionServiceABC,
settings: BotSettings
):
DiscordCommandABC.__init__(self)
self._logger = logger
self._config = config
self._message_service = message_service
self._bot = bot
self._client_utils = client_utils
self._t = translate
self._permissions = permissions
self._settings = settings
self._logger.trace(__name__, f'Loaded command service: {type(self).__name__}')
@commands.hybrid_command()
@commands.guild_only()
async def shutdown(self, ctx: Context):
self._logger.debug(__name__, f'Received command shutdown {ctx}')
if not await self._client_utils.check_if_bot_is_ready_yet_and_respond(ctx):
return
self._client_utils.received_command(ctx.guild.id)
if not self._permissions.is_member_moderator(ctx.author):
await self._message_service.send_ctx_msg(ctx, self._t.transform('common.no_permission_message'))
self._logger.trace(__name__, f'Finished shutdown command')
return
await self._client_utils.presence_game('common.presence.shutdown')
await self._message_service.send_ctx_msg(ctx, self._t.transform('modules.admin.shutdown_message'))
await asyncio.sleep(self._settings.wait_for_shutdown)
await self._bot.stop_async()
self._logger.trace(__name__, f'Finished shutdown command')

26
kdb-bot/src/modules/auto_role/__init__.py Normal file
View File

@@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
"""
bot Keksdose bot
~~~~~~~~~~~~~~~~~~~
Discord bot for the Keksdose discord Server
:copyright: (c) 2022 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'modules.auto_role'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2022 sh-edraft.de'
__version__ = '0.2.3'
from collections import namedtuple
# imports:
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='0', minor='2', micro='3')

46
kdb-bot/src/modules/auto_role/auto-role.json Normal file
View File

@@ -0,0 +1,46 @@
{
"ProjectSettings": {
"Name": "auto-role",
"Version": {
"Major": "0",
"Minor": "0",
"Micro": "0"
},
"Author": "",
"AuthorEmail": "",
"Description": "",
"LongDescription": "",
"URL": "",
"CopyrightDate": "",
"CopyrightName": "",
"LicenseName": "",
"LicenseDescription": "",
"Dependencies": [
"cpl-core>=2022.10.0.post5"
],
"DevDependencies": [
"cpl-cli==2022.10.0"
],
"PythonVersion": ">=3.10.4",
"PythonPath": {
"linux": ""
},
"Classifiers": []
},
"BuildSettings": {
"ProjectType": "library",
"SourcePath": "",
"OutputPath": "../../dist",
"Main": "admin.main",
"EntryPoint": "admin",
"IncludePackageData": false,
"Included": [],
"Excluded": [
"*/__pycache__",
"*/logs",
"*/tests"
],
"PackageData": {},
"ProjectReferences": []
}
}

29
kdb-bot/src/modules/auto_role/auto_role_module.py Normal file
View File

@@ -0,0 +1,29 @@
from cpl_core.configuration import ConfigurationABC
from cpl_core.dependency_injection import ServiceCollectionABC
from cpl_core.environment import ApplicationEnvironmentABC
from cpl_discord.discord_event_types_enum import DiscordEventTypesEnum
from cpl_discord.service.discord_collection_abc import DiscordCollectionABC
from bot_core.abc.module_abc import ModuleABC
from bot_core.configuration.feature_flags_enum import FeatureFlagsEnum
from modules.auto_role.command.auto_role_group import AutoRoleGroup
from modules.auto_role.events.auto_role_on_raw_reaction_add import AutoRoleOnRawReactionAddEvent
from modules.auto_role.events.auto_role_on_raw_reaction_remove import AutoRoleOnRawReactionRemoveEvent
from modules.auto_role.helper.reaction_handler import ReactionHandler
class AutoRoleModule(ModuleABC):
def __init__(self, dc: DiscordCollectionABC):
ModuleABC.__init__(self, dc, FeatureFlagsEnum.auto_role_module)
def configure_configuration(self, config: ConfigurationABC, env: ApplicationEnvironmentABC):
pass
def configure_services(self, services: ServiceCollectionABC, env: ApplicationEnvironmentABC):
services.add_transient(ReactionHandler)
# commands
self._dc.add_command(AutoRoleGroup)
# events
self._dc.add_event(DiscordEventTypesEnum.on_raw_reaction_add.value, AutoRoleOnRawReactionAddEvent)
self._dc.add_event(DiscordEventTypesEnum.on_raw_reaction_remove.value, AutoRoleOnRawReactionRemoveEvent)

26
kdb-bot/src/modules/auto_role/command/__init__.py Normal file
View File

@@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
"""
bot Keksdose bot
~~~~~~~~~~~~~~~~~~~
Discord bot for the Keksdose discord Server
:copyright: (c) 2022 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'modules.auto_role.command'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2022 sh-edraft.de'
__version__ = '0.2.3'
from collections import namedtuple
# imports:
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='0', minor='2', micro='3')

304
kdb-bot/src/modules/auto_role/command/auto_role_group.py Normal file
View File

@@ -0,0 +1,304 @@
from typing import List as TList, Optional
import discord
from cpl_core.database.context import DatabaseContextABC
from cpl_discord.command import DiscordCommandABC
from cpl_discord.container import TextChannel
from cpl_discord.service import DiscordBotServiceABC
from cpl_query.extension import List
from cpl_translation import TranslatePipe
from discord import app_commands
from discord.ext import commands
from discord.ext.commands import Context
from bot_core.abc.client_utils_service_abc import ClientUtilsServiceABC
from bot_core.abc.message_service_abc import MessageServiceABC
from bot_core.logging.command_logger import CommandLogger
from bot_data.abc.auto_role_repository_abc import AutoRoleRepositoryABC
from bot_data.abc.server_repository_abc import ServerRepositoryABC
from bot_data.model.auto_role import AutoRole
from bot_data.model.auto_role_rule import AutoRoleRule
from modules.permission.abc.permission_service_abc import PermissionServiceABC
class AutoRoleGroup(DiscordCommandABC):
def __init__(
self,
logger: CommandLogger,
message_service: MessageServiceABC,
bot: DiscordBotServiceABC,
client_utils: ClientUtilsServiceABC,
translate: TranslatePipe,
servers: ServerRepositoryABC,
auto_roles: AutoRoleRepositoryABC,
db_context: DatabaseContextABC,
permission_service: PermissionServiceABC,
):
DiscordCommandABC.__init__(self)
self._logger = logger
self._message_service = message_service
self._bot = bot
self._client_utils = client_utils
self._t = translate
self._servers = servers
self._auto_roles = auto_roles
self._db_context = db_context
self._permissions = permission_service
self._logger.trace(__name__, f'Loaded command service: {type(self).__name__}')
@commands.hybrid_group(name="auto-role")
@commands.guild_only()
async def auto_role(self, ctx: Context):
pass
@auto_role.command(alias='auto-roles')
@commands.guild_only()
async def list(self, ctx: Context, wait: int = None):
self._logger.debug(__name__, f'Received command auto-role list {ctx}')
if not await self._client_utils.check_if_bot_is_ready_yet_and_respond(ctx):
return
if not self._permissions.is_member_moderator(ctx.author):
await self._message_service.send_ctx_msg(ctx, self._t.transform('common.no_permission_message'))
self._logger.trace(__name__, f'Finished command auto-role list')
return
embed = discord.Embed(
title=self._t.transform('modules.auto_role.list.title'),
description=self._t.transform('modules.auto_role.list.description'),
color=int('ef9d0d', 16)
)
auto_roles = self._auto_roles.get_auto_roles()
if auto_roles.count() < 1:
await self._message_service.send_ctx_msg(ctx, self._t.transform('modules.auto_role.error.nothing_found'))
self._logger.trace(__name__, f'Finished command auto-role list')
return
auto_role_id = ''
message_id = ''
for auto_role in auto_roles:
auto_role_id += f'\n{auto_role.auto_role_id}'
message_id += f'\n{auto_role.discord_message_id}'
embed.add_field(name=self._t.transform('modules.auto_role.list.auto_role_id'), value=auto_role_id, inline=True)
embed.add_field(name=self._t.transform('modules.auto_role.list.message_id'), value=message_id, inline=True)
await self._message_service.send_ctx_msg(ctx, embed, wait_before_delete=wait)
self._logger.trace(__name__, f'Finished command auto-role list')
@auto_role.command()
@commands.guild_only()
async def add(self, ctx: Context, channel: discord.TextChannel, message_id: str):
self._logger.debug(__name__, f'Received command auto-role add {ctx} {message_id}')
if not await self._client_utils.check_if_bot_is_ready_yet_and_respond(ctx):
return
if not self._permissions.is_member_moderator(ctx.author):
await self._message_service.send_ctx_msg(ctx, self._t.transform('common.no_permission_message'))
self._logger.trace(__name__, f'Finished command auto-role add')
return
message = List(discord.Message, [message async for message in channel.history(limit=50)]).where(lambda m: m.id == int(message_id)).single_or_default()
if message is None:
self._logger.debug(__name__, f'Message with id {message_id} not found in {channel.name}')
await self._message_service.send_ctx_msg(ctx, self._t.transform('modules.auto_role.add.error.not_found').format(message_id, channel.name))
self._logger.trace(__name__, f'Finished command auto-role add')
return
if self._auto_roles.find_auto_role_by_message_id(int(message_id)) is not None:
self._logger.debug(__name__, f'auto-role for message {message_id} already exists')
await self._message_service.send_ctx_msg(ctx, self._t.transform('modules.auto_role.add.error.already_exists').format(message_id))
self._logger.trace(__name__, f'Finished command auto-role add')
return
server_id = self._servers.get_server_by_discord_id(ctx.guild.id).server_id
self._auto_roles.add_auto_role(AutoRole(server_id, int(message_id)))
self._db_context.save_changes()
self._logger.info(__name__, f'Saved auto-role for message {message_id} at server {server_id}')
await self._message_service.send_ctx_msg(ctx, self._t.transform('modules.auto_role.add.success').format(message_id))
self._logger.trace(__name__, f'Finished command auto-role add')
@add.autocomplete('message_id')
async def add_autocomplete(self, interaction: discord.Interaction, current: str) -> TList[app_commands.Choice[str]]:
channel = discord.utils.get(interaction.guild.text_channels, id=interaction.channel_id)
try:
channel_from_data = interaction.data['options'][0]['options'][0]['value']
found_channel = discord.utils.get(interaction.guild.text_channels, id=int(channel_from_data))
if found_channel is not None:
channel = found_channel
finally:
pass
messages = [message async for message in channel.history(limit=10)]
return [app_commands.Choice(name=f'{message.author}@{message.created_at}', value=str(message.id)) for message in messages if current in str(message.id)]
@auto_role.command()
@commands.guild_only()
async def remove(self, ctx: Context, auto_role: int):
self._logger.debug(__name__, f'Received command auto-role remove {ctx} {auto_role}')
if not await self._client_utils.check_if_bot_is_ready_yet_and_respond(ctx):
return
if not self._permissions.is_member_moderator(ctx.author):
await self._message_service.send_ctx_msg(ctx, self._t.transform('common.no_permission_message'))
self._logger.trace(__name__, f'Finished command auto-role remove')
return
auto_role_from_db = self._auto_roles.find_auto_role_by_id(auto_role)
if auto_role_from_db is None:
self._logger.debug(__name__, f'auto-role {auto_role} not found')
await self._message_service.send_ctx_msg(ctx, self._t.transform('modules.auto_role.remove.error.not_found').format(auto_role))
self._logger.trace(__name__, f'Finished command auto-role remove')
return
for rule in self._auto_roles.get_auto_role_rules_by_auto_role_id(auto_role_from_db.auto_role_id):
self._auto_roles.delete_auto_role_rule(rule)
self._logger.info(__name__, f'Removed auto-role rule {rule.role_id}')
self._auto_roles.delete_auto_role(auto_role_from_db)
self._db_context.save_changes()
self._logger.info(__name__, f'Removed auto-role {auto_role}')
await self._message_service.send_ctx_msg(ctx, self._t.transform('modules.auto_role.remove.success').format(auto_role))
self._logger.trace(__name__, f'Finished command auto-role remove')
@remove.autocomplete('auto_role')
async def remove_autocomplete(self, interaction: discord.Interaction, current: str) -> TList[app_commands.Choice[str]]:
auto_roles = self._auto_roles.get_auto_roles().select(lambda x: x.auto_role_id)
return [app_commands.Choice(name=auto_role, value=auto_role) for auto_role in auto_roles]
@auto_role.group()
@commands.guild_only()
async def rule(self, ctx: Context):
pass
@rule.command(alias='rules')
@commands.guild_only()
async def list(self, ctx: Context, auto_role: int, wait: int = None):
self._logger.debug(__name__, f'Received command auto-role rule list {ctx}')
if not await self._client_utils.check_if_bot_is_ready_yet_and_respond(ctx):
return
if not self._permissions.is_member_moderator(ctx.author):
await self._message_service.send_ctx_msg(ctx, self._t.transform('common.no_permission_message'))
self._logger.trace(__name__, f'Finished command auto-role rule list')
return
embed = discord.Embed(
title=self._t.transform('modules.auto_role.list.title'),
description=self._t.transform('modules.auto_role.list.description'),
color=int('ef9d0d', 16)
)
rules = self._auto_roles.get_auto_role_rules_by_auto_role_id(auto_role)
if rules.count() < 1:
await self._message_service.send_ctx_msg(ctx, self._t.transform('modules.auto_role.rule.error.id_not_found'))
self._logger.trace(__name__, f'Finished command auto-role rule list')
return
auto_role_rule_id = ''
emoji = ''
role = ''
for rule in rules:
auto_role_rule_id += f'\n{rule.role_id}'
emoji += f'\n{rule.emoji_name}'
role += f'\n{ctx.guild.get_role(rule.role_id)}'
embed.add_field(name=self._t.transform('modules.auto_role.rule.list.auto_role_rule_id'), value=auto_role_rule_id, inline=True)
embed.add_field(name=self._t.transform('modules.auto_role.rule.list.emoji'), value=emoji, inline=True)
embed.add_field(name=self._t.transform('modules.auto_role.rule.list.role'), value=role, inline=True)
await self._message_service.send_ctx_msg(ctx, embed, wait_before_delete=wait)
self._logger.trace(__name__, f'Finished command auto-role rule list')
@list.autocomplete('auto_role')
async def list_autocomplete(self, interaction: discord.Interaction, current: str) -> TList[app_commands.Choice[str]]:
auto_roles = self._auto_roles.get_auto_roles().select(lambda x: x.auto_role_id)
return [app_commands.Choice(name=auto_role, value=auto_role) for auto_role in auto_roles]
@rule.command()
@commands.guild_only()
async def add(self, ctx: Context, auto_role: int, emoji_name: str, role_id: str):
self._logger.debug(__name__, f'Received command auto-role add {ctx} {auto_role}')
if not await self._client_utils.check_if_bot_is_ready_yet_and_respond(ctx):
return
if not self._permissions.is_member_moderator(ctx.author):
await self._message_service.send_ctx_msg(ctx, self._t.transform('common.no_permission_message'))
self._logger.trace(__name__, f'Finished command auto-role rule add')
return
emoji = discord.utils.get(self._bot.emojis, name=emoji_name)
if emoji is None:
self._logger.debug(__name__, f'auto-role rule add emoji {emoji_name} not found')
await self._message_service.send_ctx_msg(ctx, self._t.transform('modules.auto_role.rule.add.error.emoji_not_found').format(emoji_name, auto_role))
self._logger.trace(__name__, f'Finished command auto-role add')
return
role = ctx.guild.get_role(int(role_id))
if role is None:
self._logger.debug(__name__, f'auto-role rule add role {role_id} not found')
await self._message_service.send_ctx_msg(ctx, self._t.transform('modules.auto_role.rule.add.error.role_not_found').format(role_id, auto_role))
self._logger.trace(__name__, f'Finished command auto-role add')
return
auto_role_from_db = self._auto_roles.get_auto_role_by_id(auto_role)
if auto_role_from_db is None:
await self._message_service.send_ctx_msg(ctx, self._t.transform('modules.auto_role.remove.error.not_found').format(auto_role))
self._logger.trace(__name__, f'Finished command auto-role rule add')
return
if self._auto_roles.get_auto_role_rules_by_auto_role_id(auto_role).where(lambda r: r.emoji_name == emoji.name and int(role_id) == role.id).count() > 0:
await self._message_service.send_ctx_msg(ctx, self._t.transform('modules.auto_role.add.error.already_exists').format(auto_role))
self._logger.trace(__name__, f'Finished command auto-role rule add')
return
self._auto_roles.add_auto_role_rule(AutoRoleRule(auto_role, emoji_name, int(role_id)))
self._db_context.save_changes()
await self._message_service.send_ctx_msg(ctx, self._t.transform('modules.auto_role.rule.add.success').format(emoji, role.name, auto_role))
self._logger.trace(__name__, f'Finished command auto-role rule add')
@add.autocomplete('auto_role')
async def add_autocomplete(self, interaction: discord.Interaction, current: str) -> TList[app_commands.Choice[str]]:
auto_roles = self._auto_roles.get_auto_roles().select(lambda x: x.auto_role_id)
return [app_commands.Choice(name=auto_role, value=auto_role) for auto_role in auto_roles]
@add.autocomplete('emoji_name')
async def add_autocomplete(self, interaction: discord.Interaction, current: str) -> TList[app_commands.Choice[str]]:
emojis = interaction.guild.emojis
return [app_commands.Choice(name=emoji.name, value=emoji.name) for emoji in emojis]
@add.autocomplete('role_id')
async def rule_add_autocomplete(self, interaction: discord.Interaction, current: str) -> TList[app_commands.Choice[str]]:
roles = interaction.guild.roles
return [app_commands.Choice(name=role.name, value=str(role.id)) for role in roles]
@rule.command()
@commands.guild_only()
async def remove(self, ctx: Context, auto_role_rule: int):
self._logger.debug(__name__, f'Received command auto-role remove {ctx} {auto_role_rule}')
if not await self._client_utils.check_if_bot_is_ready_yet_and_respond(ctx):
return
if not self._permissions.is_member_moderator(ctx.author):
await self._message_service.send_ctx_msg(ctx, self._t.transform('common.no_permission_message'))
self._logger.trace(__name__, f'Finished command auto-role remove')
return
auto_role_from_db = self._auto_roles.get_auto_role_rule_by_id(auto_role_rule)
if auto_role_from_db is None:
self._logger.debug(__name__, f'auto-role rule {auto_role_rule} not found')
await self._message_service.send_ctx_msg(ctx, self._t.transform('modules.auto_role.remove.error.not_found').format(auto_role_rule))
self._logger.trace(__name__, f'Finished command auto-role rule remove')
return
self._auto_roles.delete_auto_role_rule(auto_role_from_db)
self._db_context.save_changes()
self._logger.info(__name__, f'Removed auto-role rule {auto_role_rule}')
await self._message_service.send_ctx_msg(ctx, self._t.transform('modules.auto_role.rule.remove.success').format(auto_role_rule))
self._logger.trace(__name__, f'Finished command auto-role remove')
@remove.autocomplete('auto_role_rule')
async def remove_autocomplete(self, interaction: discord.Interaction, current: str) -> TList[app_commands.Choice[str]]:
rules = self._auto_roles.get_auto_role_rules()
return [app_commands.Choice(name=f'{rule.auto_role_rule_id} {rule.emoji_name} {interaction.guild.get_role(int(rule.role_id))}', value=rule.auto_role_rule_id) for rule in
rules]

26
kdb-bot/src/modules/auto_role/events/__init__.py Normal file
View File

@@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
"""
bot Keksdose bot
~~~~~~~~~~~~~~~~~~~
Discord bot for the Keksdose discord Server
:copyright: (c) 2022 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'modules.auto_role.events'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2022 sh-edraft.de'
__version__ = '0.2.3'
from collections import namedtuple
# imports:
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='0', minor='2', micro='3')

34
kdb-bot/src/modules/auto_role/events/auto_role_on_raw_reaction_add.py Normal file
View File

@@ -0,0 +1,34 @@
from cpl_core.logging import LoggerABC
from cpl_discord.events.on_raw_reaction_add_abc import OnRawReactionAddABC
from cpl_discord.service import DiscordBotServiceABC
from discord import RawReactionActionEvent
from bot_data.abc.auto_role_repository_abc import AutoRoleRepositoryABC
from bot_data.abc.server_repository_abc import ServerRepositoryABC
from modules.auto_role.helper.reaction_handler import ReactionHandler
class AutoRoleOnRawReactionAddEvent(OnRawReactionAddABC):
def __init__(
self,
logger: LoggerABC,
bot: DiscordBotServiceABC,
servers: ServerRepositoryABC,
auto_roles: AutoRoleRepositoryABC,
reaction_handler: ReactionHandler
):
OnRawReactionAddABC.__init__(self)
self._logger = logger
self._bot = bot
self._servers = servers
self._auto_roles = auto_roles
self._reaction_handler = reaction_handler
async def on_raw_reaction_add(self, payload: RawReactionActionEvent):
self._logger.debug(__name__, f'Module {type(self)} started')
await self._reaction_handler.handle(payload, 'add')
self._logger.debug(__name__, f'Module {type(self)} stopped')

34
kdb-bot/src/modules/auto_role/events/auto_role_on_raw_reaction_remove.py Normal file
View File

@@ -0,0 +1,34 @@
from cpl_core.logging import LoggerABC
from cpl_discord.events.on_raw_reaction_remove_abc import OnRawReactionRemoveABC
from cpl_discord.service import DiscordBotServiceABC
from discord import RawReactionActionEvent
from bot_data.abc.auto_role_repository_abc import AutoRoleRepositoryABC
from bot_data.abc.server_repository_abc import ServerRepositoryABC
from modules.auto_role.helper.reaction_handler import ReactionHandler
class AutoRoleOnRawReactionRemoveEvent(OnRawReactionRemoveABC):
def __init__(
self,
logger: LoggerABC,
bot: DiscordBotServiceABC,
servers: ServerRepositoryABC,
auto_roles: AutoRoleRepositoryABC,
reaction_handler: ReactionHandler
):
OnRawReactionRemoveABC.__init__(self)
self._logger = logger
self._bot = bot
self._servers = servers
self._auto_roles = auto_roles
self._reaction_handler = reaction_handler
async def on_raw_reaction_remove(self, payload: RawReactionActionEvent):
self._logger.debug(__name__, f'Module {type(self)} started')
await self._reaction_handler.handle(payload, 'remove')
self._logger.debug(__name__, f'Module {type(self)} stopped')

26
kdb-bot/src/modules/auto_role/helper/__init__.py Normal file
View File

@@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
"""
bot Keksdose bot
~~~~~~~~~~~~~~~~~~~
Discord bot for the Keksdose discord Server
:copyright: (c) 2022 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'modules.auto_role.helper'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2022 sh-edraft.de'
__version__ = '0.2.3'
from collections import namedtuple
# imports:
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='0', minor='2', micro='3')

70
kdb-bot/src/modules/auto_role/helper/reaction_handler.py Normal file
View File

@@ -0,0 +1,70 @@
import discord
from cpl_core.logging import LoggerABC
from cpl_discord.service import DiscordBotServiceABC
from cpl_query.extension import List
from discord import RawReactionActionEvent
from bot_data.abc.auto_role_repository_abc import AutoRoleRepositoryABC
from bot_data.abc.server_repository_abc import ServerRepositoryABC
from bot_data.model.auto_role import AutoRole
from bot_data.model.auto_role_rule import AutoRoleRule
class ReactionHandler:
def __init__(
self,
logger: LoggerABC,
bot: DiscordBotServiceABC,
servers: ServerRepositoryABC,
auto_roles: AutoRoleRepositoryABC
):
self._logger = logger
self._bot = bot
self._servers = servers
self._auto_roles = auto_roles
self._message_ids = self._auto_roles.get_auto_roles().select(lambda x: x.discord_message_id)
self._roles = self._auto_roles.get_auto_roles()
async def handle(self, payload: RawReactionActionEvent, r_type=None) -> None:
self._logger.trace(__name__, f'Handle reaction {payload} {r_type}')
if payload.message_id not in self._message_ids:
self._logger.debug(__name__, f'Message not in auto-roles - skipping')
return
guild = self._bot.get_guild(payload.guild_id)
user = guild.get_member(payload.user_id)
if user is None:
self._logger.warn(__name__, f'User {payload.user_id} in {guild.name} not found - skipping')
return
if user.bot:
self._logger.debug(__name__, f'User is bot - skipping')
return
emoji = payload.emoji.name
auto_role: AutoRole = self._roles.where(lambda x: x.discord_message_id == payload.message_id).single_or_default()
if auto_role is None:
self._logger.debug(__name__, f'auto-role for message not found - skipping')
return
rules: List[AutoRoleRule] = self._auto_roles.get_auto_role_rules_by_auto_role_id(auto_role.auto_role_id)
if rules.count() == 0:
self._logger.debug(__name__, f'auto-role rules not found - skipping')
return
for rule in rules:
if emoji != rule.emoji_name:
continue
if r_type == 'add':
role = guild.get_role(rule.role_id)
self._logger.debug(__name__, f'Assign role {role.name} to {user.name}')
await user.add_roles(role)
elif r_type == 'remove':
role = guild.get_role(rule.role_id)
self._logger.debug(__name__, f'Remove role {role.name} to {user.name}')
await user.remove_roles(role)
else:
self._logger.warn(__name__, f'Invalid reaction type {r_type}')

26
kdb-bot/src/modules/base/__init__.py Normal file
View File

@@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
"""
bot Keksdose bot
~~~~~~~~~~~~~~~~~~~
Discord bot for the Keksdose discord Server
:copyright: (c) 2022 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'modules.base'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2022 sh-edraft.de'
__version__ = '0.2.3'
from collections import namedtuple
# imports
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='0', minor='2', micro='3')

26
kdb-bot/src/modules/base/abc/__init__.py Normal file
View File

@@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
"""
bot Keksdose bot
~~~~~~~~~~~~~~~~~~~
Discord bot for the Keksdose discord Server
:copyright: (c) 2022 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'modules.base.abc'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2022 sh-edraft.de'
__version__ = '0.2.3'
from collections import namedtuple
# imports:
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='0', minor='2', micro='3')

12
kdb-bot/src/modules/base/abc/base_helper_abc.py Normal file
View File

@@ -0,0 +1,12 @@
from abc import abstractmethod, ABC
from modules.base.configuration.base_server_settings import BaseServerSettings
class BaseHelperABC(ABC):
def __init__(self):
ABC.__init__(self)
@abstractmethod
def get_config(self, g_id: int) -> BaseServerSettings: pass

46
kdb-bot/src/modules/base/base.json Normal file
View File

@@ -0,0 +1,46 @@
{
"ProjectSettings": {
"Name": "base",
"Version": {
"Major": "0",
"Minor": "0",
"Micro": "0"
},
"Author": "",
"AuthorEmail": "",
"Description": "",
"LongDescription": "",
"URL": "",
"CopyrightDate": "",
"CopyrightName": "",
"LicenseName": "",
"LicenseDescription": "",
"Dependencies": [
"cpl-core>=2022.10.0.post2"
],
"DevDependencies": [
"cpl-cli==2022.10.0"
],
"PythonVersion": ">=3.10.4",
"PythonPath": {
"linux": ""
},
"Classifiers": []
},
"BuildSettings": {
"ProjectType": "library",
"SourcePath": "",
"OutputPath": "../../dist",
"Main": "base.main",
"EntryPoint": "base",
"IncludePackageData": false,
"Included": [],
"Excluded": [
"*/__pycache__",
"*/logs",
"*/tests"
],
"PackageData": {},
"ProjectReferences": []
}
}

44
kdb-bot/src/modules/base/base_module.py Normal file
View File

@@ -0,0 +1,44 @@
from cpl_core.configuration import ConfigurationABC
from cpl_core.dependency_injection import ServiceCollectionABC
from cpl_core.environment import ApplicationEnvironmentABC
from cpl_discord.discord_event_types_enum import DiscordEventTypesEnum
from cpl_discord.service.discord_collection_abc import DiscordCollectionABC
from bot_core.abc.module_abc import ModuleABC
from bot_core.configuration.feature_flags_enum import FeatureFlagsEnum
from modules.base.abc.base_helper_abc import BaseHelperABC
from modules.base.command.afk_command import AFKCommand
from modules.base.command.help_command import HelpCommand
from modules.base.command.info_command import InfoCommand
from modules.base.command.ping_command import PingCommand
from modules.base.events.base_on_command_error_event import BaseOnCommandErrorEvent
from modules.base.events.base_on_command_event import BaseOnCommandEvent
from modules.base.events.base_on_member_join_event import BaseOnMemberJoinEvent
from modules.base.events.base_on_member_remove_event import BaseOnMemberRemoveEvent
from modules.base.events.base_on_message_event import BaseOnMessageEvent
from modules.base.events.base_on_voice_state_update_event import BaseOnVoiceStateUpdateEvent
from modules.base.service.base_helper_service import BaseHelperService
class BaseModule(ModuleABC):
def __init__(self, dc: DiscordCollectionABC):
ModuleABC.__init__(self, dc, FeatureFlagsEnum.base_module)
def configure_configuration(self, config: ConfigurationABC, env: ApplicationEnvironmentABC):
pass
def configure_services(self, services: ServiceCollectionABC, env: ApplicationEnvironmentABC):
services.add_transient(BaseHelperABC, BaseHelperService)
# commands
self._dc.add_command(AFKCommand)
self._dc.add_command(HelpCommand)
self._dc.add_command(InfoCommand)
self._dc.add_command(PingCommand)
# events
self._dc.add_event(DiscordEventTypesEnum.on_command.value, BaseOnCommandEvent)
self._dc.add_event(DiscordEventTypesEnum.on_command_error.value, BaseOnCommandErrorEvent)
self._dc.add_event(DiscordEventTypesEnum.on_member_join.value, BaseOnMemberJoinEvent)
self._dc.add_event(DiscordEventTypesEnum.on_member_remove.value, BaseOnMemberRemoveEvent)
self._dc.add_event(DiscordEventTypesEnum.on_message.value, BaseOnMessageEvent)
self._dc.add_event(DiscordEventTypesEnum.on_voice_state_update.value, BaseOnVoiceStateUpdateEvent)

26
kdb-bot/src/modules/base/command/__init__.py Normal file
View File

@@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
"""
bot Keksdose bot
~~~~~~~~~~~~~~~~~~~
Discord bot for the Keksdose discord Server
:copyright: (c) 2022 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'modules.base.command'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2022 sh-edraft.de'
__version__ = '0.2.3'
from collections import namedtuple
# imports
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='0', minor='2', micro='3')

60
kdb-bot/src/modules/base/command/afk_command.py Normal file
View File

@@ -0,0 +1,60 @@
from cpl_core.configuration import ConfigurationABC
from cpl_discord.command import DiscordCommandABC
from cpl_discord.service import DiscordBotServiceABC
from cpl_translation import TranslatePipe
from discord import VoiceChannel
from discord.ext import commands
from discord.ext.commands import Context
from bot_core.abc.client_utils_service_abc import ClientUtilsServiceABC
from bot_core.abc.message_service_abc import MessageServiceABC
from bot_core.logging.command_logger import CommandLogger
from modules.base.configuration.base_server_settings import BaseServerSettings
class AFKCommand(DiscordCommandABC):
def __init__(
self,
logger: CommandLogger,
config: ConfigurationABC,
message_service: MessageServiceABC,
bot: DiscordBotServiceABC,
client_utils: ClientUtilsServiceABC,
translate: TranslatePipe
):
DiscordCommandABC.__init__(self)
self._logger = logger
self._config = config
self._message_service = message_service
self._bot = bot
self._client_utils = client_utils
self._t = translate
self._logger.trace(__name__, f'Loaded command service: {type(self).__name__}')
@commands.hybrid_command()
@commands.guild_only()
async def afk(self, ctx: Context):
self._logger.debug(__name__, f'Received command afk {ctx}')
if not await self._client_utils.check_if_bot_is_ready_yet_and_respond(ctx):
return
self._client_utils.received_command(ctx.guild.id)
settings: BaseServerSettings = self._config.get_configuration(f'BaseServerSettings_{ctx.guild.id}')
if ctx.author.voice is None or ctx.author.voice.channel is None:
await self._message_service.send_ctx_msg(ctx, self._t.transform('modules.base.afk_command_channel_missing_message'))
self._logger.trace(__name__, f'Finished afk command')
return
self._bot.loop.create_task(self._message_service.send_ctx_msg(ctx, self._t.transform('modules.base.afk_command_move_message')))
channel: VoiceChannel = ctx.guild.get_channel(settings.afk_command_channel_id)
try:
await ctx.author.move_to(channel)
self._client_utils.moved_user(ctx.guild.id)
except Exception as e:
self._logger.error(__name__, f'Cannot move user {ctx.author.id} to channel {ctx.channel.id}', e)
await self._message_service.send_ctx_msg(ctx, self._t.transform('common.no_permission_message'))
self._logger.trace(__name__, f'Finished afk command')

52
kdb-bot/src/modules/base/command/help_command.py Normal file
View File

@@ -0,0 +1,52 @@
from typing import List
import discord
from cpl_core.configuration import ConfigurationABC
from cpl_discord.command import DiscordCommandABC
from cpl_discord.service import DiscordBotServiceABC
from discord import app_commands
from discord.ext import commands
from discord.ext.commands import Context
from bot_core.abc.client_utils_service_abc import ClientUtilsServiceABC
from bot_core.abc.message_service_abc import MessageServiceABC
from bot_core.logging.command_logger import CommandLogger
from modules.base.configuration.base_server_settings import BaseServerSettings
class HelpCommand(DiscordCommandABC):
def __init__(
self,
config: ConfigurationABC,
logger: CommandLogger,
message_service: MessageServiceABC,
bot: DiscordBotServiceABC,
client_utils: ClientUtilsServiceABC
):
DiscordCommandABC.__init__(self)
self._config = config
self._logger = logger
self._message_service = message_service
self._bot = bot
self._client_utils = client_utils
self._logger.trace(__name__, f'Loaded command service: {type(self).__name__}')
@commands.hybrid_command()
@commands.guild_only()
async def help(self, ctx: Context, persistent_flag: str = None):
self._logger.debug(__name__, f'Received command help {ctx}:{persistent_flag}')
if not await self._client_utils.check_if_bot_is_ready_yet_and_respond(ctx):
return
self._client_utils.received_command(ctx.guild.id)
settings: BaseServerSettings = self._config.get_configuration(f'BaseServerSettings_{ctx.guild.id}')
is_persistent = persistent_flag == '--stay'
await self._message_service.send_ctx_msg(ctx, settings.help_command_reference_url, is_persistent=is_persistent)
self._logger.trace(__name__, f'Finished help command')
@help.autocomplete('persistent_flag')
async def help_autocomplete(self, interaction: discord.Interaction, current: str) -> List[app_commands.Choice[str]]:
flags = ['--stay']
return [app_commands.Choice(name=key, value=key) for key in flags]

69
kdb-bot/src/modules/base/command/info_command.py Normal file
View File

@@ -0,0 +1,69 @@
from datetime import datetime
import discord
from cpl_core.configuration import ConfigurationABC
from cpl_discord.command import DiscordCommandABC
from cpl_discord.service import DiscordBotServiceABC
from cpl_translation import TranslatePipe
from discord.ext import commands
from discord.ext.commands import Context
import bot
from bot_core.abc.client_utils_service_abc import ClientUtilsServiceABC
from bot_core.abc.message_service_abc import MessageServiceABC
from bot_core.logging.command_logger import CommandLogger
class InfoCommand(DiscordCommandABC):
def __init__(
self,
config: ConfigurationABC,
logger: CommandLogger,
message_service: MessageServiceABC,
bot: DiscordBotServiceABC,
client_utils: ClientUtilsServiceABC,
translate: TranslatePipe
):
DiscordCommandABC.__init__(self)
self._config = config
self._logger = logger
self._message_service = message_service
self._bot = bot
self._client_utils = client_utils
self._t = translate
self._logger.trace(__name__, f'Loaded command service: {type(self).__name__}')
@commands.hybrid_command()
@commands.guild_only()
async def info(self, ctx: Context, *, wait: int = None):
self._logger.debug(__name__, f'Received command info {ctx},{wait}')
if not await self._client_utils.check_if_bot_is_ready_yet_and_respond(ctx):
return
self._client_utils.received_command(ctx.guild.id)
client = self._client_utils.get_client(self._bot.user.id, ctx.guild.id)
embed = discord.Embed(
title=self._t.transform('modules.base.info.title'),
description=self._t.transform('modules.base.info.description'),
color=int('ef9d0d', 16)
)
embed.add_field(name=self._t.transform('modules.base.info.fields.version'), value=bot.__version__)
start_time = self._config.get_configuration('Bot_StartTime')
ontime = round((datetime.now() - datetime.strptime(start_time, '%Y-%m-%d %H:%M:%S.%f')).total_seconds() / 3600, 2)
embed.add_field(name=self._t.transform('modules.base.info.fields.ontime'), value=f'{ontime}h')
embed.add_field(name=self._t.transform('modules.base.info.fields.sent_message_count'), value=client.sent_message_count, inline=False)
embed.add_field(name=self._t.transform('modules.base.info.fields.received_message_count'), value=client.received_message_count)
embed.add_field(name=self._t.transform('modules.base.info.fields.deleted_message_count'), value=client.deleted_message_count, inline=False)
embed.add_field(name=self._t.transform('modules.base.info.fields.received_command_count'), value=client.received_command_count)
embed.add_field(name=self._t.transform('modules.base.info.fields.moved_users_count'), value=client.moved_users_count)
from bot.module_list import ModuleList
modules = ModuleList.get_modules()
modules = modules.select(lambda x: x.__name__.replace('Module', ''))
embed.add_field(name=self._t.transform('modules.base.info.fields.modules'), value='\n'.join(modules), inline=False)
await self._message_service.send_ctx_msg(ctx, embed, wait_before_delete=wait)
self._logger.trace(__name__, f'Finished info command')

40
kdb-bot/src/modules/base/command/ping_command.py Normal file
View File

@@ -0,0 +1,40 @@
from cpl_discord.command import DiscordCommandABC
from cpl_discord.service import DiscordBotServiceABC
from cpl_translation import TranslatePipe
from discord.ext import commands
from discord.ext.commands import Context
from bot_core.abc.client_utils_service_abc import ClientUtilsServiceABC
from bot_core.abc.message_service_abc import MessageServiceABC
from bot_core.logging.command_logger import CommandLogger
class PingCommand(DiscordCommandABC):
def __init__(
self,
logger: CommandLogger,
message_service: MessageServiceABC,
bot: DiscordBotServiceABC,
client_utils: ClientUtilsServiceABC,
translate: TranslatePipe
):
DiscordCommandABC.__init__(self)
self._logger = logger
self._message_service = message_service
self._bot = bot
self._client_utils = client_utils
self._t = translate
self._logger.trace(__name__, f'Loaded command service: {type(self).__name__}')
@commands.hybrid_command()
@commands.guild_only()
async def ping(self, ctx: Context):
self._logger.debug(__name__, f'Received command ping {ctx}')
if not await self._client_utils.check_if_bot_is_ready_yet_and_respond(ctx):
return
self._client_utils.received_command(ctx.guild.id)
await self._message_service.send_ctx_msg(ctx, self._t.transform('modules.base.pong'))
self._logger.trace(__name__, f'Finished ping command')

26
kdb-bot/src/modules/base/configuration/__init__.py Normal file
View File

@@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
"""
bot Keksdose bot
~~~~~~~~~~~~~~~~~~~
Discord bot for the Keksdose discord Server
:copyright: (c) 2022 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'modules.base.configuration'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2022 sh-edraft.de'
__version__ = '0.2.3'
from collections import namedtuple
# imports:
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='0', minor='2', micro='3')

61
kdb-bot/src/modules/base/configuration/base_server_settings.py Normal file
View File

@@ -0,0 +1,61 @@
import traceback
from cpl_core.configuration.configuration_model_abc import ConfigurationModelABC
from cpl_core.console import Console
from cpl_query.extension import List
class BaseServerSettings(ConfigurationModelABC):
def __init__(self):
ConfigurationModelABC.__init__(self)
self._id: int = 0
self._max_voice_state_hours: int = 0
self._xp_per_message: int = 0
self._xp_per_ontime_hour: int = 0
self._afk_channel_ids: List[int] = List(int)
self._afk_command_channel_id: int = 0
self._help_command_reference_url: str = ''
@property
def id(self) -> int:
return self._id
@property
def max_voice_state_hours(self) -> int:
return self._max_voice_state_hours
@property
def xp_per_message(self) -> int:
return self._xp_per_message
@property
def xp_per_ontime_hour(self) -> int:
return self._xp_per_ontime_hour
@property
def afk_channel_ids(self) -> List[int]:
return self._afk_channel_ids
@property
def afk_command_channel_id(self) -> int:
return self._afk_command_channel_id
@property
def help_command_reference_url(self) -> str:
return self._help_command_reference_url
def from_dict(self, settings: dict):
try:
self._id = int(settings['Id'])
self._max_voice_state_hours = int(settings['MaxVoiceStateHours'])
self._xp_per_message = int(settings['XpPerMessage'])
self._xp_per_ontime_hour = int(settings['XpPerOntimeHour'])
for index in settings['AFKChannelIds']:
self._afk_channel_ids.append(int(index))
self._afk_command_channel_id = settings['AFKCommandChannelId']
self._help_command_reference_url = settings['HelpCommandReferenceUrl']
except Exception as e:
Console.error(f'[ ERROR ] [ {__name__} ]: Reading error in {type(self).__name__} settings')
Console.error(f'[ EXCEPTION ] [ {__name__} ]: {e} -> {traceback.format_exc()}')

32
kdb-bot/src/modules/base/configuration/base_settings.py Normal file
View File

@@ -0,0 +1,32 @@
import traceback
from cpl_core.configuration import ConfigurationModelABC
from cpl_core.console import Console
from cpl_query.extension import List
from modules.base.configuration.base_server_settings import BaseServerSettings
class BaseSettings(ConfigurationModelABC):
def __init__(self):
ConfigurationModelABC.__init__(self)
self._servers: List[BaseServerSettings] = List()
@property
def servers(self) -> List[BaseServerSettings]:
return self._servers
def from_dict(self, settings: dict):
try:
servers = List(BaseServerSettings)
for s in settings:
st = BaseServerSettings()
settings[s]['Id'] = s
st.from_dict(settings[s])
servers.append(st)
self._servers = servers
except Exception as e:
Console.error(f'[ ERROR ] [ {__name__} ]: Reading error in {type(self).__name__} settings')
Console.error(f'[ EXCEPTION ] [ {__name__} ]: {e} -> {traceback.format_exc()}')

26
kdb-bot/src/modules/base/events/__init__.py Normal file
View File

@@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
"""
bot Keksdose bot
~~~~~~~~~~~~~~~~~~~
Discord bot for the Keksdose discord Server
:copyright: (c) 2022 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'modules.base.events'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2022 sh-edraft.de'
__version__ = '0.2.3'
from collections import namedtuple
# imports:
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='0', minor='2', micro='3')

146
kdb-bot/src/modules/base/events/base_on_command_error_event.py Normal file
View File

@@ -0,0 +1,146 @@
import datetime
import traceback
import uuid
from cpl_core.time import TimeFormatSettings
from cpl_discord.service import DiscordBotServiceABC
from cpl_translation import TranslatePipe
from discord.ext import commands
from discord.ext.commands import Context, CommandError
from cpl_core.logging import LoggerABC
from cpl_discord.events.on_command_error_abc import OnCommandErrorABC
from bot_core.abc.message_service_abc import MessageServiceABC
from bot_core.configuration.bot_settings import BotSettings
class BaseOnCommandErrorEvent(OnCommandErrorABC):
def __init__(
self,
logger: LoggerABC,
bot: DiscordBotServiceABC,
messenger: MessageServiceABC,
bot_settings: BotSettings,
time_format_settings: TimeFormatSettings,
translate: TranslatePipe
):
OnCommandErrorABC.__init__(self)
self._logger = logger
self._bot = bot
self._messenger = messenger
self._bot_settings = bot_settings
self._time_format_settings = time_format_settings
self._t = translate
async def on_command_error(self, ctx: Context, error: CommandError):
error = getattr(error, 'original', error)
uid = uuid.uuid4()
self._logger.error(__name__, f'Got error: {type(error).__name__} UID: {uid}')
if isinstance(error, commands.MissingRequiredArgument):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.missing_required_argument'), without_tracking=True)
elif isinstance(error, commands.ArgumentParsingError):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.argument_parsing_error'), without_tracking=True)
elif isinstance(error, commands.UnexpectedQuoteError):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.unexpected_quote_error'), without_tracking=True)
elif isinstance(error, commands.InvalidEndOfQuotedStringError):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.invalid_end_of_quoted_string_error'), without_tracking=True)
elif isinstance(error, commands.ExpectedClosingQuoteError):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.expected_closing_quote_error'), without_tracking=True)
elif isinstance(error, commands.BadArgument):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.bad_argument'), without_tracking=True)
elif isinstance(error, commands.BadUnionArgument):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.bad_union_argument'), without_tracking=True)
elif isinstance(error, commands.PrivateMessageOnly):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.private_message_only'), without_tracking=True)
elif isinstance(error, commands.NoPrivateMessage):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.no_private_message'), without_tracking=True)
elif isinstance(error, commands.CheckFailure):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.check_failure'), without_tracking=True)
elif isinstance(error, commands.CheckAnyFailure):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.check_any_failure'), without_tracking=True)
elif isinstance(error, commands.CommandNotFound):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.command_not_found'), without_tracking=True)
elif isinstance(error, commands.DisabledCommand):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.disabled_command'), without_tracking=True)
elif isinstance(error, commands.CommandInvokeError):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.command_invoke_error'), without_tracking=True)
elif isinstance(error, commands.TooManyArguments):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.too_many_arguments'), without_tracking=True)
elif isinstance(error, commands.UserInputError):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.user_input_error'), without_tracking=True)
elif isinstance(error, commands.CommandOnCooldown):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.command_on_cooldown'), without_tracking=True)
elif isinstance(error, commands.MaxConcurrencyReached):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.max_concurrency_reached'), without_tracking=True)
elif isinstance(error, commands.NotOwner):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.not_owner'), without_tracking=True)
elif isinstance(error, commands.MissingPermissions):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.missing_permissions'), without_tracking=True)
elif isinstance(error, commands.BotMissingPermissions):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.bot_missing_permissions'), without_tracking=True)
elif isinstance(error, commands.MissingRole):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.missing_role'), without_tracking=True)
elif isinstance(error, commands.BotMissingRole):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.bot_missing_role'), without_tracking=True)
elif isinstance(error, commands.MissingAnyRole):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.missing_any_role'), without_tracking=True)
elif isinstance(error, commands.BotMissingAnyRole):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.bot_missing_any_role'), without_tracking=True)
elif isinstance(error, commands.NSFWChannelRequired):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.nsfw_channel_required'), without_tracking=True)
elif isinstance(error, commands.ExtensionError):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.extension_error'), without_tracking=True)
elif isinstance(error, commands.ExtensionAlreadyLoaded):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.extension_already_loaded'), without_tracking=True)
elif isinstance(error, commands.ExtensionNotLoaded):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.extension_not_loaded'), without_tracking=True)
elif isinstance(error, commands.NoEntryPointError):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.no_entry_point_error'), without_tracking=True)
elif isinstance(error, commands.ExtensionFailed):
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.extension_failed'), without_tracking=True)
else:
await self._messenger.send_ctx_msg(ctx, self._t.transform('common.errors.command_error'), without_tracking=True)
message = self._t.transform('modules.base.technician_command_error_message').format(
ctx.command,
ctx.author,
error,
datetime.datetime.now().strftime(self._time_format_settings.date_time_format),
uid
)
for t in self._bot_settings.technicians:
member = self._bot.get_user(t)
await self._messenger.send_dm_message(message, member, without_tracking=True)

90
kdb-bot/src/modules/base/events/base_on_command_event.py Normal file
View File

@@ -0,0 +1,90 @@
from typing import Optional
import discord
from cpl_core.database.context import DatabaseContextABC
from cpl_core.time import TimeFormatSettings
from cpl_discord.events import OnCommandABC
from cpl_discord.service import DiscordBotServiceABC
from cpl_translation import TranslatePipe
from discord.ext.commands import Context
from bot_core.abc.message_service_abc import MessageServiceABC
from bot_core.configuration.bot_settings import BotSettings
from bot_core.logging.command_logger import CommandLogger
from bot_data.abc.client_repository_abc import ClientRepositoryABC
from bot_data.abc.server_repository_abc import ServerRepositoryABC
from bot_data.abc.user_repository_abc import UserRepositoryABC
from bot_data.model.user import User
from modules.base.abc.base_helper_abc import BaseHelperABC
from modules.base.configuration.base_server_settings import BaseServerSettings
class BaseOnCommandEvent(OnCommandABC):
def __init__(
self,
logger: CommandLogger,
bot: DiscordBotServiceABC,
messenger: MessageServiceABC,
bot_settings: BotSettings,
time_format_settings: TimeFormatSettings,
translate: TranslatePipe,
bhs: BaseHelperABC,
db: DatabaseContextABC,
users: UserRepositoryABC,
clients: ClientRepositoryABC,
servers: ServerRepositoryABC,
):
OnCommandABC.__init__(self)
self._logger = logger
self._bot = bot
self._messenger = messenger
self._bot_settings = bot_settings
self._time_format_settings = time_format_settings
self._t = translate
self._base_helper = bhs
self._db = db
self._users = users
self._clients = clients
self._servers = servers
def _append_received_command_count(self, g_id: int):
try:
self._clients.append_received_command_count(self._bot.user.id, g_id, 1)
self._db.save_changes()
except Exception as e:
self._logger.error(__name__, f'Cannot edit client {self._bot.user.id}@{g_id}', e)
def _handle_message_for_xp(self, message: discord.Message):
dc_user_id = message.author.id
try:
server = self._servers.get_server_by_discord_id(message.guild.id)
except Exception as e:
self._logger.error(__name__, f'Cannot get server {message.guild.id}', e)
return
user: Optional[User] = None
try:
user = self._users.get_user_by_discord_id_and_server_id(dc_user_id, server.server_id)
except Exception as e:
self._logger.error(__name__, f'Cannot get user {dc_user_id}', e)
return
if user is None:
self._logger.error(__name__, f'User not found {dc_user_id}')
return
settings: BaseServerSettings = self._base_helper.get_config(message.guild.id)
old_xp = user.xp
user.xp += settings.xp_per_message
self._users.update_user(user)
self._db.save_changes()
self._logger.debug(__name__, f'User {user} sent message. xp: from {old_xp} to {user.xp}')
async def on_command(self, ctx: Context):
self._logger.debug(__name__, f'Module {type(self)} started')
self._logger.info(__name__, f'Received command: {ctx.command} from {ctx.channel}')
if ctx is None or ctx.guild is None:
return
self._append_received_command_count(ctx.guild.id)

97
kdb-bot/src/modules/base/events/base_on_member_join_event.py Normal file
View File

@@ -0,0 +1,97 @@
from datetime import datetime
from typing import Union
import discord
from cpl_core.configuration import ConfigurationABC
from cpl_core.database.context import DatabaseContextABC
from cpl_core.logging import LoggerABC
from cpl_discord.events import OnMemberJoinABC
from cpl_translation import TranslatePipe
from bot_core.abc.message_service_abc import MessageServiceABC
from bot_data.abc.known_user_repository_abc import KnownUserRepositoryABC
from bot_data.abc.server_repository_abc import ServerRepositoryABC
from bot_data.abc.user_joined_server_repository_abc import UserJoinedServerRepositoryABC
from bot_data.abc.user_repository_abc import UserRepositoryABC
from bot_data.model.known_user import KnownUser
from bot_data.model.user import User
from bot_data.model.user_joined_server import UserJoinedServer
from modules.base.abc.base_helper_abc import BaseHelperABC
from modules.base.configuration.base_server_settings import BaseServerSettings
from modules.permission.abc.permission_service_abc import PermissionServiceABC
class BaseOnMemberJoinEvent(OnMemberJoinABC):
def __init__(
self,
config: ConfigurationABC,
logger: LoggerABC,
base_helper: BaseHelperABC,
messenger: MessageServiceABC,
permissions: PermissionServiceABC,
db: DatabaseContextABC,
known_users: KnownUserRepositoryABC,
users: UserRepositoryABC,
servers: ServerRepositoryABC,
user_joins: UserJoinedServerRepositoryABC,
translate: TranslatePipe
):
OnMemberJoinABC.__init__(self)
self._config = config
self._logger = logger
self._base_helper = base_helper
self._messenger = messenger
self._permission_service = permissions
self._db = db
self._known_users = known_users
self._users = users
self._servers = servers
self._user_joins = user_joins
self._t = translate
def _check_for_known_user(self, member: Union[discord.User, discord.Member]):
self._logger.debug(__name__, f'Check if user is already known {member}')
try:
user = self._known_users.find_user_by_discord_id(member.id)
if user is not None:
return
self._logger.debug(__name__, f'Add user: {member.id}')
self._known_users.add_user(KnownUser(member.id))
self._db.save_changes()
except Exception as e:
self._logger.error(__name__, f'Cannot get user {member.id}', e)
async def _add_if_not_exists_user_async(self, member: Union[discord.User, discord.Member]):
self._logger.debug(__name__, f'Check if user exists {member}')
settings: BaseServerSettings = self._base_helper.get_config(member.guild.id)
await self._messenger.send_dm_message(self._t.transform('modules.base.welcome_message').format(member.guild.name), member)
for admin in self._permission_service.get_admins(member.guild.id):
await self._messenger.send_dm_message(self._t.transform('modules.base.welcome_message_for_team').format(member.name), admin)
for moderator in self._permission_service.get_moderators(member.guild.id):
await self._messenger.send_dm_message(self._t.transform('modules.base.welcome_message_for_team').format(member.name), moderator)
try:
server = self._servers.get_server_by_discord_id(member.guild.id)
user = self._users.find_user_by_discord_id_and_server_id(member.id, server.server_id)
if user is not None:
self._user_joins.add_user_joined_server(UserJoinedServer(user, datetime.now()))
return
self._logger.debug(__name__, f'Add user: {member.id}')
self._users.add_user(User(member.id, 0, server))
self._db.save_changes()
user = self._users.get_user_by_discord_id_and_server_id(member.id, server.server_id)
self._user_joins.add_user_joined_server(UserJoinedServer(user, datetime.now()))
self._db.save_changes()
except Exception as e:
self._logger.error(__name__, f'Cannot get user {member.id}', e)
async def on_member_join(self, member: discord.Member):
self._logger.debug(__name__, f'Module {type(self)} started')
self._check_for_known_user(member)
await self._add_if_not_exists_user_async(member)

64
kdb-bot/src/modules/base/events/base_on_member_remove_event.py Normal file
View File

@@ -0,0 +1,64 @@
from datetime import datetime
from typing import Union
import discord
from cpl_core.database.context import DatabaseContextABC
from cpl_core.logging import LoggerABC
from cpl_discord.events import OnMemberRemoveABC
from cpl_translation import TranslatePipe
from bot_core.abc.message_service_abc import MessageServiceABC
from bot_data.abc.server_repository_abc import ServerRepositoryABC
from bot_data.abc.user_joined_server_repository_abc import UserJoinedServerRepositoryABC
from bot_data.abc.user_repository_abc import UserRepositoryABC
from modules.base.abc.base_helper_abc import BaseHelperABC
from modules.base.configuration.base_server_settings import BaseServerSettings
class BaseOnMemberRemoveEvent(OnMemberRemoveABC):
def __init__(
self,
logger: LoggerABC,
base_helper: BaseHelperABC,
db: DatabaseContextABC,
messenger: MessageServiceABC,
users: UserRepositoryABC,
servers: ServerRepositoryABC,
user_joins: UserJoinedServerRepositoryABC,
translate: TranslatePipe
):
OnMemberRemoveABC.__init__(self)
self._logger = logger
self._base_helper = base_helper
self._db = db
self._messenger = messenger
self._users = users
self._servers = servers
self._user_joins = user_joins
self._t = translate
async def _remove_user(self, member: Union[discord.User, discord.Member]):
self._logger.debug(__name__, f'Remove user {member}')
settings: BaseServerSettings = self._base_helper.get_config(member.guild.id)
await self._messenger.send_dm_message(self._t.transform('modules.base.goodbye_message'), member)
try:
server = self._servers.get_server_by_discord_id(member.guild.id)
user = self._users.find_user_by_discord_id_and_server_id(member.id, server.server_id)
if user is None:
self._logger.error(__name__, f'Cannot find user {member}')
return
join = self._user_joins.get_active_user_joined_server_by_user_id(user.user_id)
join.leaved_on = datetime.now()
self._user_joins.update_user_joined_server(join)
self._db.save_changes()
except Exception as e:
self._logger.error(__name__, f'Cannot get user {member.id}', e)
async def on_member_remove(self, member: discord.Member):
self._logger.debug(__name__, f'Module {type(self)} started')
await self._remove_user(member)

81
kdb-bot/src/modules/base/events/base_on_message_event.py Normal file
View File

@@ -0,0 +1,81 @@
from typing import Optional
import discord
from cpl_core.database.context import DatabaseContextABC
from cpl_discord.events import OnMessageABC
from cpl_discord.service import DiscordBotServiceABC
from bot_core.helper.log_message_helper import LogMessageHelper
from bot_core.logging.message_logger import MessageLogger
from bot_data.abc.client_repository_abc import ClientRepositoryABC
from bot_data.abc.server_repository_abc import ServerRepositoryABC
from bot_data.abc.user_repository_abc import UserRepositoryABC
from bot_data.model.user import User
from modules.base.abc.base_helper_abc import BaseHelperABC
from modules.base.configuration.base_server_settings import BaseServerSettings
class BaseOnMessageEvent(OnMessageABC):
def __init__(
self,
logger: MessageLogger,
bhs: BaseHelperABC,
db: DatabaseContextABC,
bot: DiscordBotServiceABC,
users: UserRepositoryABC,
clients: ClientRepositoryABC,
servers: ServerRepositoryABC,
):
OnMessageABC.__init__(self)
self._logger = logger
self._base_helper = bhs
self._db = db
self._bot = bot
self._users = users
self._clients = clients
self._servers = servers
def _append_received_message_count(self, g_id: int):
try:
self._clients.append_received_message_count(self._bot.user.id, g_id, 1)
self._db.save_changes()
except Exception as e:
self._logger.error(__name__, f'Cannot edit client {self._bot.user.id}@{g_id}', e)
def _handle_message_for_xp(self, message: discord.Message):
dc_user_id = message.author.id
try:
server = self._servers.get_server_by_discord_id(message.guild.id)
except Exception as e:
self._logger.error(__name__, f'Cannot get server {message.guild.id}', e)
return
user: Optional[User] = None
try:
user = self._users.find_user_by_discord_id_and_server_id(dc_user_id, server.server_id)
except Exception as e:
self._logger.error(__name__, f'Cannot get user {dc_user_id}', e)
return
if user is None:
self._logger.error(__name__, f'User not found {dc_user_id}')
return
settings: BaseServerSettings = self._base_helper.get_config(message.guild.id)
old_xp = user.xp
user.xp += settings.xp_per_message
self._users.update_user(user)
self._db.save_changes()
self._logger.debug(__name__, f'User {user} sent message. xp: from {old_xp} to {user.xp}')
async def on_message(self, message: discord.Message):
self._logger.debug(__name__, f'Module {type(self)} started')
self._logger.info(__name__, f'Received message: {LogMessageHelper.get_log_string(message)}')
if message is None or message.guild is None:
return
self._append_received_message_count(message.guild.id)
if not message.author.bot:
self._handle_message_for_xp(message)

118
kdb-bot/src/modules/base/events/base_on_voice_state_update_event.py Normal file
View File

@@ -0,0 +1,118 @@
from datetime import datetime
from typing import Optional
import discord
from cpl_core.configuration import ConfigurationABC
from cpl_core.database.context import DatabaseContextABC
from cpl_core.logging import LoggerABC
from cpl_discord.events import OnVoiceStateUpdateABC
from bot_data.abc.known_user_repository_abc import KnownUserRepositoryABC
from bot_data.abc.server_repository_abc import ServerRepositoryABC
from bot_data.abc.user_joined_server_repository_abc import UserJoinedServerRepositoryABC
from bot_data.abc.user_joined_voice_channel_abc import UserJoinedVoiceChannelRepositoryABC
from bot_data.abc.user_repository_abc import UserRepositoryABC
from bot_data.model.server import Server
from bot_data.model.user import User
from bot_data.model.user_joined_voice_channel import UserJoinedVoiceChannel
from modules.base.abc.base_helper_abc import BaseHelperABC
from modules.base.configuration.base_server_settings import BaseServerSettings
class BaseOnVoiceStateUpdateEvent(OnVoiceStateUpdateABC):
def __init__(
self,
config: ConfigurationABC,
logger: LoggerABC,
base_helper: BaseHelperABC,
servers: ServerRepositoryABC,
known_users: KnownUserRepositoryABC,
users: UserRepositoryABC,
user_joins: UserJoinedServerRepositoryABC,
user_joins_vc: UserJoinedVoiceChannelRepositoryABC,
db: DatabaseContextABC,
):
OnVoiceStateUpdateABC.__init__(self)
self._config = config
self._logger = logger
self._base_helper = base_helper
self._servers = servers
self._known_users = known_users
self._users = users
self._user_joins = user_joins
self._user_joins_vc = user_joins_vc
self._db = db
self._logger.info(__name__, f'Module {type(self)} loaded')
def _update_voice_state(self, joined: bool, dc_user_id: int, dc_channel_id: int, server: Server):
user: Optional[User] = None
try:
user = self._users.get_user_by_discord_id_and_server_id(dc_user_id, server.server_id)
except Exception as e:
self._logger.error(__name__, f'Cannot get user {dc_user_id}', e)
return
if user is None:
self._logger.error(__name__, f'User not found {dc_user_id}')
return
try:
if joined:
join = UserJoinedVoiceChannel(user, dc_channel_id, datetime.now())
self._user_joins_vc.add_user_joined_voice_channel(join)
self._db.save_changes()
return
settings: BaseServerSettings = self._base_helper.get_config(server.discord_server_id)
join = self._user_joins_vc.get_active_user_joined_voice_channel_by_user_id(user.user_id)
join.leaved_on = datetime.now()
# ontime as hours
ontime = round((join.leaved_on - join.joined_on).total_seconds() / 3600, 2)
old_xp = user.xp
user.xp += round(ontime * settings.xp_per_ontime_hour)
self._user_joins_vc.update_user_joined_voice_channel(join)
self._users.update_user(user)
self._db.save_changes()
self._logger.debug(__name__, f'User {user} leaved_on {join.leaved_on}. Ontime: {ontime}h | xp: from {old_xp} to {user.xp}')
except Exception as e:
self._logger.error(__name__, f'Ontime validation failed', e)
async def on_voice_state_update(self, member: discord.Member, before: discord.VoiceState, after: discord.VoiceState):
self._logger.debug(__name__, f'Module {type(self)} started')
self._logger.trace(__name__, f'Detected on_voice_state_update {member.id} from {before} to {after}')
settings: BaseServerSettings = self._base_helper.get_config(member.guild.id)
server = self._servers.get_server_by_discord_id(member.guild.id)
try:
# join
if before.channel is None and after.channel is not None and after.channel.id not in settings.afk_channel_ids:
self._logger.trace(__name__, f'User {member.id} joined {after.channel}')
self._update_voice_state(True, member.id, after.channel.id, server)
# leave
elif before.channel is not None and after.channel is None and before.channel.id not in settings.afk_channel_ids:
self._logger.trace(__name__, f'User {member.id} left {before.channel}')
self._update_voice_state(False, member.id, before.channel.id, server)
# channel to channel
elif before.channel is not None and after.channel is not None:
# joined
if before.channel.id in settings.afk_channel_ids and after.channel.id not in settings.afk_channel_ids:
self._logger.trace(__name__, f'User {member.id} joined {after.channel}')
self._update_voice_state(True, member.id, after.channel.id, server)
# left
elif after.channel.id in settings.afk_channel_ids and before.channel.id not in settings.afk_channel_ids:
self._logger.trace(__name__, f'User {member.id} left {before.channel}')
self._update_voice_state(False, member.id, before.channel.id, server)
else:
self._logger.trace(__name__, f'User {member.id} switched to {after.channel}')
except Exception as e:
self._logger.error(__name__, f'Cannot handle voice state for user {member.id}', e)

26
kdb-bot/src/modules/base/service/__init__.py Normal file
View File

@@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
"""
bot Keksdose bot
~~~~~~~~~~~~~~~~~~~
Discord bot for the Keksdose discord Server
:copyright: (c) 2022 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'modules.base.service'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2022 sh-edraft.de'
__version__ = '0.2.3'
from collections import namedtuple
# imports
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='0', minor='2', micro='3')

14
kdb-bot/src/modules/base/service/base_helper_service.py Normal file
View File

@@ -0,0 +1,14 @@
from cpl_core.configuration import ConfigurationABC
from modules.base.abc.base_helper_abc import BaseHelperABC
from modules.base.configuration.base_server_settings import BaseServerSettings
class BaseHelperService(BaseHelperABC):
def __init__(self, config: ConfigurationABC):
BaseHelperABC.__init__(self)
self._config = config
def get_config(self, g_id: int) -> BaseServerSettings:
return self._config.get_configuration(f'BaseServerSettings_{g_id}')

26
kdb-bot/src/modules/boot_log/__init__.py Normal file
View File

@@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
"""
bot Keksdose bot
~~~~~~~~~~~~~~~~~~~
Discord bot for the Keksdose discord Server
:copyright: (c) 2022 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'modules.boot_log'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2022 sh-edraft.de'
__version__ = '0.2.3'
from collections import namedtuple
# imports:
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='0', minor='2', micro='3')

46
kdb-bot/src/modules/boot_log/boot-log.json Normal file
View File

@@ -0,0 +1,46 @@
{
"ProjectSettings": {
"Name": "boot-log",
"Version": {
"Major": "0",
"Minor": "0",
"Micro": "0"
},
"Author": "",
"AuthorEmail": "",
"Description": "",
"LongDescription": "",
"URL": "",
"CopyrightDate": "",
"CopyrightName": "",
"LicenseName": "",
"LicenseDescription": "",
"Dependencies": [
"cpl-core>=2022.10.0.post2"
],
"DevDependencies": [
"cpl-cli==2022.10.0"
],
"PythonVersion": ">=3.10.4",
"PythonPath": {
"linux": ""
},
"Classifiers": []
},
"BuildSettings": {
"ProjectType": "library",
"SourcePath": "",
"OutputPath": "../../dist",
"Main": "boot_log.main",
"EntryPoint": "boot-log",
"IncludePackageData": false,
"Included": [],
"Excluded": [
"*/__pycache__",
"*/logs",
"*/tests"
],
"PackageData": {},
"ProjectReferences": []
}
}

23
kdb-bot/src/modules/boot_log/boot_log_extension.py Normal file
View File

@@ -0,0 +1,23 @@
from datetime import datetime
from cpl_core.application.application_extension_abc import ApplicationExtensionABC
from cpl_core.configuration import ConfigurationABC
from cpl_core.dependency_injection import ServiceProviderABC
from cpl_core.logging import LoggerABC
from bot_core.configuration.feature_flags_enum import FeatureFlagsEnum
from bot_core.configuration.feature_flags_settings import FeatureFlagsSettings
class BootLogExtension(ApplicationExtensionABC):
def __init__(self):
pass
async def run(self, config: ConfigurationABC, services: ServiceProviderABC):
feature_flags: FeatureFlagsSettings = config.get_configuration(FeatureFlagsSettings)
if not feature_flags.get_flag(FeatureFlagsEnum.boot_log_module):
return
logger: LoggerABC = services.get_service(LoggerABC)
logger.debug(__name__, 'BootLog extension started')
config.add_configuration('Bot_StartTime', str(datetime.now()))

23
kdb-bot/src/modules/boot_log/boot_log_module.py Normal file
View File

@@ -0,0 +1,23 @@
from cpl_core.configuration import ConfigurationABC
from cpl_core.dependency_injection import ServiceCollectionABC
from cpl_core.environment import ApplicationEnvironmentABC
from cpl_discord.discord_event_types_enum import DiscordEventTypesEnum
from cpl_discord.service.discord_collection_abc import DiscordCollectionABC
from bot_core.abc.module_abc import ModuleABC
from bot_core.configuration.feature_flags_enum import FeatureFlagsEnum
from modules.boot_log.boot_log_on_ready_event import BootLogOnReadyEvent
class BootLogModule(ModuleABC):
def __init__(self, dc: DiscordCollectionABC):
ModuleABC.__init__(self, dc, FeatureFlagsEnum.boot_log_module)
def configure_configuration(self, config: ConfigurationABC, env: ApplicationEnvironmentABC):
pass
def configure_services(self, services: ServiceCollectionABC, env: ApplicationEnvironmentABC):
# commands
# events
self._dc.add_event(DiscordEventTypesEnum.on_ready.value, BootLogOnReadyEvent)

78
kdb-bot/src/modules/boot_log/boot_log_on_ready_event.py Normal file
View File

@@ -0,0 +1,78 @@
from datetime import datetime
from cpl_core.configuration import ConfigurationABC
from cpl_core.logging import LoggerABC
from cpl_discord.events import OnReadyABC
from cpl_discord.service import DiscordBotServiceABC
from cpl_translation import TranslatePipe
from discord import guild
from bot_core.abc.message_service_abc import MessageServiceABC
from bot_core.configuration.server_settings import ServerSettings
from modules.boot_log.configuration.boot_log_server_settings import BootLogServerSettings
class BootLogOnReadyEvent(OnReadyABC):
def __init__(
self,
config: ConfigurationABC,
logger: LoggerABC,
bot: DiscordBotServiceABC,
message_service: MessageServiceABC,
translate: TranslatePipe
):
OnReadyABC.__init__(self)
self._config = config
self._logger = logger
self._bot = bot
self._message_service = message_service
self._t = translate
self._logger.info(__name__, f'Module {type(self)} loaded')
async def on_ready(self):
self._logger.debug(__name__, f'Module {type(self)} started')
try:
start_time = self._config.get_configuration('Bot_StartTime')
init_time = round((datetime.now() - datetime.strptime(start_time, '%Y-%m-%d %H:%M:%S.%f')).total_seconds(), 2)
self._config.add_configuration('InitTime', str(init_time))
self._logger.debug(__name__, f'Bot Init time: {init_time}s')
# print warning if initialisation took too long
if init_time >= 30:
self._logger.warn(
__name__, 'It takes too long to start the bot!')
# print error if initialisation took way too long
elif init_time >= 90:
self._logger.error(
__name__, 'It takes far too long to start the bot!!!')
except Exception as e:
self._logger.error(__name__, 'Init time calculation failed', e)
return
for g in self._bot.guilds:
g: guild = g
self._logger.debug(__name__, f'Server detected: {g.id}')
server_settings: ServerSettings = self._config.get_configuration(f'ServerSettings_{g.id}')
if server_settings is None:
self._logger.error(__name__, f'BootLog settings for server {g.id} not found!')
return
module_settings: BootLogServerSettings = self._config.get_configuration(f'BootLogServerSettings_{g.id}')
if module_settings is None:
self._logger.error(__name__, f'Config {type(self).__name__}_{g.id} not found!')
return
await self._message_service.send_channel_message(
self._bot.get_channel(
module_settings.login_message_channel_id
),
self._t.transform('modules.boot_log.login_message').format(init_time)
)
self._config.add_configuration('IS_READY', 'true')
self._logger.info(__name__, 'Bot is ready')
self._logger.trace(__name__, f'Module {type(self)} stopped')

26
kdb-bot/src/modules/boot_log/configuration/__init__.py Normal file
View File

@@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
"""
bot Keksdose bot
~~~~~~~~~~~~~~~~~~~
Discord bot for the Keksdose discord Server
:copyright: (c) 2022 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'modules.boot_log.configuration'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2022 sh-edraft.de'
__version__ = '0.2.3'
from collections import namedtuple
# imports:
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='0', minor='2', micro='3')

29
kdb-bot/src/modules/boot_log/configuration/boot_log_server_settings.py Normal file
View File

@@ -0,0 +1,29 @@
import traceback
from cpl_core.configuration.configuration_model_abc import ConfigurationModelABC
from cpl_core.console import Console
class BootLogServerSettings(ConfigurationModelABC):
def __init__(self):
ConfigurationModelABC.__init__(self)
self._id: int = 0
self._login_message_channel_id: int = 0
@property
def id(self) -> int:
return self._id
@property
def login_message_channel_id(self) -> int:
return self._login_message_channel_id
def from_dict(self, settings: dict):
try:
self._id = int(settings['Id'])
self._login_message_channel_id = int(settings['LoginMessageChannelId'])
except Exception as e:
Console.error(f'[ ERROR ] [ {__name__} ]: Reading error in {type(self).__name__} settings')
Console.error(f'[ EXCEPTION ] [ {__name__} ]: {e} -> {traceback.format_exc()}')

32
kdb-bot/src/modules/boot_log/configuration/boot_log_settings.py Normal file
View File

@@ -0,0 +1,32 @@
import traceback
from cpl_core.configuration.configuration_model_abc import ConfigurationModelABC
from cpl_core.console import Console
from cpl_query.extension import List
from modules.boot_log.configuration.boot_log_server_settings import BootLogServerSettings
class BootLogSettings(ConfigurationModelABC):
def __init__(self):
ConfigurationModelABC.__init__(self)
self._servers: List[BootLogServerSettings] = List()
@property
def servers(self) -> List[BootLogServerSettings]:
return self._servers
def from_dict(self, settings: dict):
try:
servers = List(BootLogServerSettings)
for s in settings:
st = BootLogServerSettings()
settings[s]['Id'] = s
st.from_dict(settings[s])
servers.append(st)
self._servers = servers
except Exception as e:
Console.error(f'[ ERROR ] [ {__name__} ]: Reading error in {type(self).__name__} settings')
Console.error(f'[ EXCEPTION ] [ {__name__} ]: {e} -> {traceback.format_exc()}')

26
kdb-bot/src/modules/database/__init__.py Normal file
View File

@@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
"""
bot Keksdose bot
~~~~~~~~~~~~~~~~~~~
Discord bot for the Keksdose discord Server
:copyright: (c) 2022 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'modules.database'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2022 sh-edraft.de'
__version__ = '0.2.3'
from collections import namedtuple
# imports:
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='0', minor='2', micro='3')

46
kdb-bot/src/modules/database/database.json Normal file
View File

@@ -0,0 +1,46 @@
{
"ProjectSettings": {
"Name": "database",
"Version": {
"Major": "1",
"Minor": "0",
"Micro": "0.dev1"
},
"Author": "Sven Heidemann",
"AuthorEmail": "sven.heidemann@sh-edraft.de",
"Description": "Keksdose bot - db module",
"LongDescription": "Discord bot for the Keksdose discord Server - database module",
"URL": "https://www.sh-edraft.de",
"CopyrightDate": "2022",
"CopyrightName": "sh-edraft.de",
"LicenseName": "MIT",
"LicenseDescription": "MIT, see LICENSE for more details.",
"Dependencies": [
"cpl-core>=2022.10.0.post2"
],
"DevDependencies": [
"cpl-cli==2022.10.0"
],
"PythonVersion": ">=3.10.4",
"PythonPath": {
"linux": ""
},
"Classifiers": []
},
"BuildSettings": {
"ProjectType": "library",
"SourcePath": "",
"OutputPath": "../../dist",
"Main": "database.main",
"EntryPoint": "database",
"IncludePackageData": false,
"Included": [],
"Excluded": [
"*/__pycache__",
"*/logs",
"*/tests"
],
"PackageData": {},
"ProjectReferences": []
}
}

27
kdb-bot/src/modules/database/database_extension.py Normal file
View File

@@ -0,0 +1,27 @@
from datetime import datetime
from cpl_core.application.application_extension_abc import ApplicationExtensionABC
from cpl_core.configuration import ConfigurationABC
from cpl_core.dependency_injection import ServiceProviderABC
from cpl_core.logging import LoggerABC
from bot_core.configuration.feature_flags_enum import FeatureFlagsEnum
from bot_core.configuration.feature_flags_settings import FeatureFlagsSettings
from bot_core.logging.database_logger import DatabaseLogger
from bot_data.service.migration_service import MigrationService
class DatabaseExtension(ApplicationExtensionABC):
def __init__(self):
pass
async def run(self, config: ConfigurationABC, services: ServiceProviderABC):
feature_flags: FeatureFlagsSettings = config.get_configuration(FeatureFlagsSettings)
if not feature_flags.get_flag(FeatureFlagsEnum.data_module):
return
logger: LoggerABC = services.get_service(DatabaseLogger)
logger.debug(__name__, 'Database extension started')
config.add_configuration('Database_StartTime', str(datetime.now()))
migrations: MigrationService = services.get_service(MigrationService)
migrations.migrate()

23
kdb-bot/src/modules/database/database_module.py Normal file
View File

@@ -0,0 +1,23 @@
from cpl_core.configuration import ConfigurationABC
from cpl_core.dependency_injection import ServiceCollectionABC
from cpl_core.environment import ApplicationEnvironmentABC
from cpl_discord.discord_event_types_enum import DiscordEventTypesEnum
from cpl_discord.service.discord_collection_abc import DiscordCollectionABC
from bot_core.abc.module_abc import ModuleABC
from bot_core.configuration.feature_flags_enum import FeatureFlagsEnum
from modules.database.database_on_ready_event import DatabaseOnReadyEvent
class DatabaseModule(ModuleABC):
def __init__(self, dc: DiscordCollectionABC):
ModuleABC.__init__(self, dc, FeatureFlagsEnum.database_module)
def configure_configuration(self, config: ConfigurationABC, env: ApplicationEnvironmentABC):
pass
def configure_services(self, services: ServiceCollectionABC, env: ApplicationEnvironmentABC):
# commands
# events
self._dc.add_event(DiscordEventTypesEnum.on_ready.value, DatabaseOnReadyEvent)

314
kdb-bot/src/modules/database/database_on_ready_event.py Normal file
View File

@@ -0,0 +1,314 @@
from ctypes import Union
from datetime import datetime, timedelta
import discord
from cpl_core.configuration import ConfigurationABC
from cpl_core.database.context import DatabaseContextABC
from cpl_discord.events import OnReadyABC
from cpl_discord.service import DiscordBotServiceABC
from bot_core.logging.database_logger import DatabaseLogger
from bot_core.pipes.date_time_offset_pipe import DateTimeOffsetPipe
from bot_data.abc.client_repository_abc import ClientRepositoryABC
from bot_data.abc.known_user_repository_abc import KnownUserRepositoryABC
from bot_data.abc.user_joined_server_repository_abc import UserJoinedServerRepositoryABC
from bot_data.abc.user_joined_voice_channel_abc import UserJoinedVoiceChannelRepositoryABC
from bot_data.abc.user_repository_abc import UserRepositoryABC
from bot_data.model.client import Client
from bot_data.model.known_user import KnownUser
from bot_data.model.server import Server
from bot_data.model.user import User
from bot_data.model.user_joined_server import UserJoinedServer
from bot_data.model.user_joined_voice_channel import UserJoinedVoiceChannel
from bot_data.service.user_repository_service import ServerRepositoryABC
from modules.base.configuration.base_server_settings import BaseServerSettings
class DatabaseOnReadyEvent(OnReadyABC):
def __init__(
self,
config: ConfigurationABC,
logger: DatabaseLogger,
bot: DiscordBotServiceABC,
db_context: DatabaseContextABC,
server_repo: ServerRepositoryABC,
user_repo: UserRepositoryABC,
client_repo: ClientRepositoryABC,
known_users: KnownUserRepositoryABC,
user_joins: UserJoinedServerRepositoryABC,
user_joins_vc: UserJoinedVoiceChannelRepositoryABC,
dtp: DateTimeOffsetPipe
):
self._config = config
self._logger = logger
self._bot = bot
self._db_context = db_context
self._servers = server_repo
self._users = user_repo
self._clients = client_repo
self._known_users = known_users
self._user_joins = user_joins
self._user_joins_vc = user_joins_vc
self._dtp = dtp
OnReadyABC.__init__(self)
self._logger.info(__name__, f'Module {type(self)} loaded')
def _validate_init_time(self):
try:
start_time = self._config.get_configuration('Database_StartTime')
init_time = round((datetime.now() - datetime.strptime(start_time, '%Y-%m-%d %H:%M:%S.%f')).total_seconds(), 2)
self._config.add_configuration('Database_InitTime', str(init_time))
self._logger.debug(__name__, f'Database Init time: {init_time}s')
# print warning if initialisation took too long
if init_time >= 30:
self._logger.warn(
__name__, 'It takes too long to prepare the database!')
# print error if initialisation took way too long
elif init_time >= 90:
self._logger.error(
__name__, 'It takes far too long to prepare the database!!!')
except Exception as e: #
self._logger.error(__name__, 'Database init time calculation failed', e)
return
def _check_known_users(self):
self._logger.debug(__name__, f'Start checking KnownUsers table, {len(self._bot.users)}')
for u in self._bot.users:
u: discord.User = u
try:
if u.bot:
self._logger.trace(__name__, f'User {u.id} is ignored, because its a bot')
continue
user = self._known_users.find_user_by_discord_id(u.id)
if user is not None:
continue
self._logger.warn(__name__, f'Unknown user: {u.id}')
self._logger.debug(__name__, f'Add user: {u.id}')
self._known_users.add_user(KnownUser(u.id))
self._db_context.save_changes()
user = self._known_users.find_user_by_discord_id(u.id)
if user is None:
self._logger.fatal(__name__, f'Cannot add user: {u.id}')
self._logger.debug(__name__, f'Added user: {u.id}')
except Exception as e:
self._logger.error(__name__, f'Cannot get user', e)
def _check_servers(self):
self._logger.debug(__name__, f'Start checking Servers table')
for g in self._bot.guilds:
g: discord.Guild = g
try:
server = self._servers.find_server_by_discord_id(g.id)
if server is not None:
continue
self._logger.warn(__name__, f'Server not found in database: {g.id}')
self._logger.debug(__name__, f'Add server: {g.id}')
self._servers.add_server(Server(g.id))
self._db_context.save_changes()
server = self._servers.find_server_by_discord_id(g.id)
if server is None:
self._logger.fatal(__name__, f'Cannot add server: {g.id}')
self._logger.debug(__name__, f'Added server: {g.id}')
except Exception as e:
self._logger.error(__name__, f'Cannot get server', e)
results = self._servers.get_servers()
if results is None or len(results) == 0:
self._logger.error(__name__, f'Table Servers is empty!')
def _check_clients(self):
self._logger.debug(__name__, f'Start checking Clients table')
for g in self._bot.guilds:
g: discord.Guild = g
try:
server: Server = self._servers.find_server_by_discord_id(g.id)
if server is None:
self._logger.fatal(__name__, f'Server not found in database: {g.id}')
client = self._clients.find_client_by_server_id(server.server_id)
if client is not None:
continue
self._logger.warn(__name__, f'Client for server {g.id} not found in database: {self._bot.user.id}')
self._logger.debug(__name__, f'Add client: {self._bot.user.id}')
self._clients.add_client(Client(self._bot.user.id, 0, 0, 0, 0, 0, server))
self._db_context.save_changes()
client = self._clients.find_client_by_server_id(server.server_id)
if client is None:
self._logger.fatal(__name__, f'Cannot add client {self._bot.user.id} for server {g.id}')
self._logger.debug(__name__, f'Added client: {g.id}')
except Exception as e:
self._logger.error(__name__, f'Cannot get client', e)
results = self._servers.get_servers()
if results is None or len(results) == 0:
self._logger.error(__name__, f'Table Servers is empty!')
def _check_users(self):
self._logger.debug(__name__, f'Start checking Users table')
for g in self._bot.guilds:
g: discord.Guild = g
try:
server = self._servers.find_server_by_discord_id(g.id)
if server is None:
self._logger.fatal(__name__, f'Server not found in database: {g.id}')
for u in g.members:
u: Union[discord.Member, discord.User] = u
if u.bot:
self._logger.trace(__name__, f'User {u.id} is ignored, because its a bot')
continue
user = self._users.find_user_by_discord_id_and_server_id(u.id, server.server_id)
if user is not None:
continue
self._logger.warn(__name__, f'User not found in database: {u.id}')
self._logger.debug(__name__, f'Add user: {u.id}')
self._users.add_user(User(u.id, 0, server))
self._db_context.save_changes()
self._logger.debug(__name__, f'Added User: {u.id}')
except Exception as e:
self._logger.error(__name__, f'Cannot get User', e)
results = self._users.get_users()
if results is None or len(results) == 0:
self._logger.error(__name__, f'Table Users is empty!')
def _check_user_joins(self):
self._logger.debug(__name__, f'Start checking UserJoinedServers table')
for guild in self._bot.guilds:
guild: discord.Guild = guild
server = self._servers.find_server_by_discord_id(guild.id)
if server is None:
self._logger.fatal(__name__, f'Server not found in database: {guild.id}')
try:
for u in guild.members:
u: discord.User = u
if u.bot:
self._logger.trace(__name__, f'User {u.id} is ignored, because its a bot')
continue
user = self._users.find_user_by_discord_id_and_server_id(u.id, server.server_id)
if user is None:
self._logger.fatal(__name__, f'User not found in database: {u.id}')
join = self._user_joins.find_active_user_joined_server_by_user_id(user.user_id)
if join is not None:
continue
m: discord.Member = u
self._logger.warn(__name__, f'Active UserJoinedServer not found in database: {guild.id}:{u.id}@{m.joined_at}')
self._logger.debug(__name__, f'Add UserJoinedServer: {guild.id}:{u.id}@{m.joined_at}')
self._user_joins.add_user_joined_server(UserJoinedServer(user, self._dtp.transform(m.joined_at), None))
self._db_context.save_changes()
self._logger.debug(__name__, f'Added UserJoinedServer: {u.id}')
except Exception as e:
self._logger.error(__name__, f'Cannot get UserJoinedServer', e)
results = self._users.get_users()
if results is None or len(results) == 0:
self._logger.error(__name__, f'Table Users is empty!')
joins = self._user_joins.get_user_joined_servers()
for join in joins:
join: UserJoinedServer = join
if join.user.server.discord_server_id != guild.id:
continue
if join.leaved_on is not None:
continue
dc_user = guild.get_member(join.user.discord_id)
if dc_user is None:
self._logger.warn(__name__, f'User {join.user.discord_id} already left the server.')
join.leaved_on = datetime.now()
self._user_joins.update_user_joined_server(join)
self._db_context.save_changes()
def _check_user_joins_vc(self):
self._logger.debug(__name__, f'Start checking UserJoinedVoiceChannel table')
for guild in self._bot.guilds:
guild: discord.Guild = guild
server = self._servers.find_server_by_discord_id(guild.id)
if server is None:
self._logger.fatal(__name__, f'Server not found in database: {guild.id}')
try:
for member in guild.members:
if member.bot:
self._logger.trace(__name__, f'User {member.id} is ignored, because its a bot')
continue
user = self._users.find_user_by_discord_id_and_server_id(member.id, server.server_id)
if user is None:
self._logger.fatal(__name__, f'User not found in database: {member.id}')
joins = self._user_joins_vc.find_active_user_joined_voice_channels_by_user_id(user.user_id)
if joins is None or len(joins) == 0:
continue
for join in joins:
self._logger.warn(__name__, f'Active UserJoinedVoiceChannel found in database: {guild.id}:{member.id}@{join.joined_on}')
join.leaved_on = datetime.now()
settings: BaseServerSettings = self._config.get_configuration(f'BaseServerSettings_{guild.id}')
if ((join.leaved_on - join.joined_on).total_seconds() / 60 / 60) > settings.max_voice_state_hours:
join.leaved_on = join.joined_on + timedelta(hours=settings.max_voice_state_hours)
self._user_joins_vc.update_user_joined_voice_channel(join)
# todo: maybe add XP
self._db_context.save_changes()
for member in guild.members:
if member.bot:
self._logger.trace(__name__, f'User {member.id} is ignored, because its a bot')
continue
if member.voice is None:
continue
user = self._users.find_user_by_discord_id_and_server_id(member.id, server.server_id)
if user is None:
self._logger.fatal(__name__, f'User not found in database: {member.id}')
join = UserJoinedVoiceChannel(user, member.voice.channel.id, datetime.now())
self._user_joins_vc.add_user_joined_voice_channel(join)
self._db_context.save_changes()
self._logger.warn(__name__, f'VS {member.voice}')
except Exception as e:
self._logger.error(__name__, f'Cannot get UserJoinedVoiceChannel', e)
async def on_ready(self):
self._logger.debug(__name__, f'Module {type(self)} started')
self._check_known_users()
self._check_servers()
self._check_clients()
self._check_users()
self._check_user_joins()
self._check_user_joins_vc()
self._validate_init_time()
self._logger.trace(__name__, f'Module {type(self)} stopped')

26
kdb-bot/src/modules/moderator/__init__.py Normal file
View File

@@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
"""
bot Keksdose bot
~~~~~~~~~~~~~~~~~~~
Discord bot for the Keksdose discord Server
:copyright: (c) 2022 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'modules.moderator'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2022 sh-edraft.de'
__version__ = '0.2.3'
from collections import namedtuple
# imports:
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='0', minor='2', micro='3')

26
kdb-bot/src/modules/moderator/command/__init__.py Normal file
View File

@@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
"""
bot Keksdose bot
~~~~~~~~~~~~~~~~~~~
Discord bot for the Keksdose discord Server
:copyright: (c) 2022 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'modules.moderator.command'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2022 sh-edraft.de'
__version__ = '0.2.3'
from collections import namedtuple
# imports:
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='0', minor='2', micro='3')

61
kdb-bot/src/modules/moderator/command/purge_command.py Normal file
View File

@@ -0,0 +1,61 @@
import asyncio
from cpl_core.configuration import ConfigurationABC
from cpl_discord.command import DiscordCommandABC
from cpl_translation import TranslatePipe
from discord.ext import commands
from discord.ext.commands import Context
from bot_core.abc.client_utils_service_abc import ClientUtilsServiceABC
from bot_core.abc.message_service_abc import MessageServiceABC
from bot_core.configuration.server_settings import ServerSettings
from bot_core.logging.command_logger import CommandLogger
from modules.permission.abc.permission_service_abc import PermissionServiceABC
class PurgeCommand(DiscordCommandABC):
def __init__(
self,
logger: CommandLogger,
config: ConfigurationABC,
message_service: MessageServiceABC,
permissions: PermissionServiceABC,
client_utils: ClientUtilsServiceABC,
translate: TranslatePipe
):
DiscordCommandABC.__init__(self)
self._logger = logger
self._config = config
self._message_service = message_service
self._permissions = permissions
self._client_utils = client_utils
self._t = translate
self._logger.trace(__name__, f'Loaded command service: {type(self).__name__}')
@commands.hybrid_command()
@commands.guild_only()
async def purge(self, ctx: Context):
self._logger.debug(__name__, f'Received command purge {ctx}')
if not await self._client_utils.check_if_bot_is_ready_yet_and_respond(ctx):
return
self._client_utils.received_command(ctx.guild.id)
server_settings: ServerSettings = self._config.get_configuration(f'ServerSettings_{ctx.guild.id}')
if not self._permissions.is_member_moderator(ctx.author):
await self._message_service.send_ctx_msg(ctx, self._t.transform('common.no_permission_message'))
self._logger.trace(__name__, f'Finished purge command')
return
await self._message_service.send_ctx_msg(ctx, self._t.transform('modules.moderator.purge_message'))
await asyncio.sleep(server_settings.message_delete_timer)
try:
await ctx.channel.purge()
except Exception as e:
self._logger.error(__name__, f'Cannot purge channel {ctx.channel.id}', e)
await self._message_service.send_ctx_msg(ctx, self._t.transform('common.bot_has_no_permission_message'))
self._logger.trace(__name__, f'Finished purge command')

111
kdb-bot/src/modules/moderator/command/user_group.py Normal file
View File

@@ -0,0 +1,111 @@
from typing import Optional
import discord
from cpl_core.configuration import ConfigurationABC
from cpl_discord.command import DiscordCommandABC
from cpl_discord.service import DiscordBotServiceABC
from cpl_translation import TranslatePipe
from discord.ext import commands
from discord.ext.commands import Context
from bot_core.abc.client_utils_service_abc import ClientUtilsServiceABC
from bot_core.abc.message_service_abc import MessageServiceABC
from bot_core.logging.command_logger import CommandLogger
from bot_core.pipes.date_time_offset_pipe import DateTimeOffsetPipe
from bot_data.abc.server_repository_abc import ServerRepositoryABC
from bot_data.abc.user_joined_server_repository_abc import UserJoinedServerRepositoryABC
from bot_data.abc.user_repository_abc import UserRepositoryABC
from modules.permission.abc.permission_service_abc import PermissionServiceABC
class UserGroup(DiscordCommandABC):
def __init__(
self,
config: ConfigurationABC,
logger: CommandLogger,
message_service: MessageServiceABC,
bot: DiscordBotServiceABC,
client_utils: ClientUtilsServiceABC,
permissions: PermissionServiceABC,
servers: ServerRepositoryABC,
users: UserRepositoryABC,
user_joined_servers: UserJoinedServerRepositoryABC,
translate: TranslatePipe,
date: DateTimeOffsetPipe
):
DiscordCommandABC.__init__(self)
self._config = config
self._logger = logger
self._message_service = message_service
self._bot = bot
self._client_utils = client_utils
self._permissions = permissions
self._servers = servers
self._users = users
self._user_joined_servers = user_joined_servers
self._t = translate
self._date = date
self._logger.trace(__name__, f'Loaded command service: {type(self).__name__}')
@commands.hybrid_group()
@commands.guild_only()
async def user(self, ctx: Context):
pass
@user.command()
@commands.guild_only()
async def info(self, ctx: Context, member: Optional[discord.Member] = None, *, wait: int = None):
self._logger.debug(__name__, f'Received command user-info {ctx}:{member},{wait}')
if not await self._client_utils.check_if_bot_is_ready_yet_and_respond(ctx):
return
self._client_utils.received_command(ctx.guild.id)
if not self._permissions.is_member_moderator(ctx.author):
await self._message_service.send_ctx_msg(ctx, self._t.transform('common.no_permission_message'))
self._logger.trace(__name__, f'Finished purge command')
return
if member is None or not isinstance(member, discord.Member):
member = ctx.author
server = self._servers.find_server_by_discord_id(ctx.guild.id)
user = self._users.find_user_by_discord_id_and_server_id(member.id, server.server_id)
joins = self._user_joined_servers.get_user_joined_servers_by_user_id(user.user_id)
embed = discord.Embed(
title=member.name,
description=member.name,
color=int('ef9d0d', 16)
)
embed.add_field(name=self._t.transform('modules.base.user_info.fields.id'), value=member.id)
embed.add_field(name=self._t.transform('modules.base.user_info.fields.name'), value=member.name)
embed.add_field(name=self._t.transform('modules.base.user_info.fields.discord_join'), value=self._date.transform(member.created_at), inline=False)
embed.add_field(name=self._t.transform('modules.base.user_info.fields.last_join'), value=self._date.transform(member.joined_at), inline=False)
embed.add_field(name=self._t.transform('modules.base.user_info.fields.xp'), value=str(user.xp))
roles = ''
for role in member.roles:
roles += f'{role.name}\n'
embed.add_field(name=self._t.transform('modules.base.user_info.fields.roles'), value=roles, inline=False)
joins_string = ''
for join in joins:
joins_string += f'{self._date.transform(join.joined_on)}\n'
embed.add_field(name=self._t.transform('modules.base.user_info.fields.joins'), value=joins_string)
lefts_string = ''
for join in joins:
if join.leaved_on is None:
if lefts_string == '':
lefts_string = '/'
continue
lefts_string += f'{self._date.transform(join.leaved_on)}\n'
embed.add_field(name=self._t.transform('modules.base.user_info.fields.lefts'), value=lefts_string)
embed.add_field(name=self._t.transform('modules.base.user_info.fields.warnings'), value=self._t.transform('common.not_implemented_yet'), inline=False)
await self._message_service.send_ctx_msg(ctx, embed, wait_before_delete=wait)
self._logger.trace(__name__, f'Finished user-info command')

46
kdb-bot/src/modules/moderator/moderator.json Normal file
View File

@@ -0,0 +1,46 @@
{
"ProjectSettings": {
"Name": "moderator",
"Version": {
"Major": "0",
"Minor": "0",
"Micro": "0"
},
"Author": "",
"AuthorEmail": "",
"Description": "",
"LongDescription": "",
"URL": "",
"CopyrightDate": "",
"CopyrightName": "",
"LicenseName": "",
"LicenseDescription": "",
"Dependencies": [
"cpl-core>=2022.10.0.post5"
],
"DevDependencies": [
"cpl-cli==2022.10.0"
],
"PythonVersion": ">=3.10.4",
"PythonPath": {
"linux": ""
},
"Classifiers": []
},
"BuildSettings": {
"ProjectType": "library",
"SourcePath": "",
"OutputPath": "../../dist",
"Main": "moderator.main",
"EntryPoint": "moderator",
"IncludePackageData": false,
"Included": [],
"Excluded": [
"*/__pycache__",
"*/logs",
"*/tests"
],
"PackageData": {},
"ProjectReferences": []
}
}

24
kdb-bot/src/modules/moderator/moderator_module.py Normal file
View File

@@ -0,0 +1,24 @@
from cpl_core.configuration import ConfigurationABC
from cpl_core.dependency_injection import ServiceCollectionABC
from cpl_core.environment import ApplicationEnvironmentABC
from cpl_discord.service.discord_collection_abc import DiscordCollectionABC
from bot_core.abc.module_abc import ModuleABC
from bot_core.configuration.feature_flags_enum import FeatureFlagsEnum
from modules.moderator.command.purge_command import PurgeCommand
from modules.moderator.command.user_group import UserGroup
class ModeratorModule(ModuleABC):
def __init__(self, dc: DiscordCollectionABC):
ModuleABC.__init__(self, dc, FeatureFlagsEnum.moderator_module)
def configure_configuration(self, config: ConfigurationABC, env: ApplicationEnvironmentABC):
pass
def configure_services(self, services: ServiceCollectionABC, env: ApplicationEnvironmentABC):
# commands
self._dc.add_command(PurgeCommand)
self._dc.add_command(UserGroup)
# events

26
kdb-bot/src/modules/permission/__init__.py Normal file
View File

@@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
"""
bot Keksdose bot
~~~~~~~~~~~~~~~~~~~
Discord bot for the Keksdose discord Server
:copyright: (c) 2022 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'modules.permission'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2022 sh-edraft.de'
__version__ = '0.2.3'
from collections import namedtuple
# imports:
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='0', minor='2', micro='3')

26
kdb-bot/src/modules/permission/abc/__init__.py Normal file
View File

@@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
"""
bot Keksdose bot
~~~~~~~~~~~~~~~~~~~
Discord bot for the Keksdose discord Server
:copyright: (c) 2022 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'modules.permission.abc'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2022 sh-edraft.de'
__version__ = '0.2.3'
from collections import namedtuple
# imports
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='0', minor='2', micro='3')

45
kdb-bot/src/modules/permission/abc/permission_service_abc.py Normal file
View File

@@ -0,0 +1,45 @@
from abc import ABC, abstractmethod
import discord
class PermissionServiceABC(ABC):
@abstractmethod
def __init__(self): pass
@abstractmethod
def on_ready(self): pass
@abstractmethod
def on_member_update(self, before: discord.Member, after: discord.Member): pass
@abstractmethod
def get_admin_role_ids(self, g_id: int) -> list[int]: pass
@abstractmethod
def get_admin_roles(self, g_id: int) -> list[discord.Role]: pass
@abstractmethod
def get_admins(self, g_id: int) -> list[discord.Member]: pass
@abstractmethod
def get_moderator_role_ids(self, g_id: int) -> list[int]: pass
@abstractmethod
def get_moderator_roles(self, g_id: int) -> list[discord.Role]: pass
@abstractmethod
def get_moderators(self, g_id: int) -> list[discord.Member]: pass
@abstractmethod
def get_technicians(self) -> list[discord.Member]: pass
@abstractmethod
def is_member_admin(self, member: discord.Member) -> bool: pass
@abstractmethod
def is_member_moderator(self, member: discord.Member) -> bool: pass
@abstractmethod
def is_member_technician(self, member: discord.Member) -> bool: pass

26
kdb-bot/src/modules/permission/configuration/__init__.py Normal file
View File

@@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
"""
bot Keksdose bot
~~~~~~~~~~~~~~~~~~~
Discord bot for the Keksdose discord Server
:copyright: (c) 2022 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'modules.permission.configuration'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2022 sh-edraft.de'
__version__ = '0.2.3'
from collections import namedtuple
# imports
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='0', minor='2', micro='3')

38
kdb-bot/src/modules/permission/configuration/permission_server_settings.py Normal file
View File

@@ -0,0 +1,38 @@
import traceback
from cpl_core.configuration.configuration_model_abc import ConfigurationModelABC
from cpl_core.console import Console
class PermissionServerSettings(ConfigurationModelABC):
def __init__(self):
ConfigurationModelABC.__init__(self)
self._id: int = 0
self._admin_roles: list[int] = []
self._moderator_roles: list[int] = []
@property
def id(self) -> int:
return self._id
@property
def admin_roles(self) -> list[int]:
return self._admin_roles
@property
def moderator_roles(self) -> list[int]:
return self._moderator_roles
def from_dict(self, settings: dict):
try:
self._id = int(settings['Id'])
for index in settings['AdminRoleIds']:
self._admin_roles.append(int(index))
for index in settings['ModeratorRoleIds']:
self._moderator_roles.append(int(index))
except Exception as e:
Console.error(f'[ ERROR ] [ {__name__} ]: Reading error in {self.__name__} settings')
Console.error(f'[ EXCEPTION ] [ {__name__} ]: {e} -> {traceback.format_exc()}')

32
kdb-bot/src/modules/permission/configuration/permission_settings.py Normal file
View File

@@ -0,0 +1,32 @@
import traceback
from cpl_core.configuration.configuration_model_abc import ConfigurationModelABC
from cpl_core.console import Console
from cpl_query.extension import List
from modules.permission.configuration.permission_server_settings import PermissionServerSettings
class PermissionSettings(ConfigurationModelABC):
def __init__(self):
ConfigurationModelABC.__init__(self)
self._servers: List[PermissionServerSettings] = List()
@property
def servers(self) -> List[PermissionServerSettings]:
return self._servers
def from_dict(self, settings: dict):
try:
servers = List(PermissionServerSettings)
for s in settings:
st = PermissionServerSettings()
settings[s]['Id'] = s
st.from_dict(settings[s])
servers.append(st)
self._servers = servers
except Exception as e:
Console.error(f'[ ERROR ] [ {__name__} ]: Reading error in {type(self).__name__} settings')
Console.error(f'[ EXCEPTION ] [ {__name__} ]: {e} -> {traceback.format_exc()}')

26
kdb-bot/src/modules/permission/events/__init__.py Normal file
View File

@@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
"""
bot Keksdose bot
~~~~~~~~~~~~~~~~~~~
Discord bot for the Keksdose discord Server
:copyright: (c) 2022 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'modules.permission.events'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2022 sh-edraft.de'
__version__ = '0.2.3'
from collections import namedtuple
# imports:
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='0', minor='2', micro='3')

19
kdb-bot/src/modules/permission/events/permission_on_member_update_event.py Normal file
View File

@@ -0,0 +1,19 @@
import discord
from cpl_core.logging import LoggerABC
from cpl_discord.events import OnMemberUpdateABC
from modules.permission.abc.permission_service_abc import PermissionServiceABC
class PermissionOnMemberUpdateEvent(OnMemberUpdateABC):
def __init__(self, logger: LoggerABC, permission_service: PermissionServiceABC):
OnMemberUpdateABC.__init__(self)
self._logger = logger
self._permission_service = permission_service
async def on_member_update(self, before: discord.Member, after: discord.Member):
self._logger.debug(__name__, f'Module {type(self)} started')
if before.roles != after.roles:
self._permission_service.on_member_update(before, after)

16
kdb-bot/src/modules/permission/events/permission_on_ready_event.py Normal file
View File

@@ -0,0 +1,16 @@
from cpl_core.logging import LoggerABC
from cpl_discord.events import OnReadyABC
from modules.permission.abc.permission_service_abc import PermissionServiceABC
class PermissionOnReadyEvent(OnReadyABC):
def __init__(self, logger: LoggerABC, permission_service: PermissionServiceABC):
OnReadyABC.__init__(self)
self._logger = logger
self._permission_service = permission_service
async def on_ready(self):
self._logger.debug(__name__, f'Module {type(self)} started')
self._permission_service.on_ready()

46
kdb-bot/src/modules/permission/permission.json Normal file
View File

@@ -0,0 +1,46 @@
{
"ProjectSettings": {
"Name": "permission",
"Version": {
"Major": "0",
"Minor": "0",
"Micro": "0"
},
"Author": "",
"AuthorEmail": "",
"Description": "",
"LongDescription": "",
"URL": "",
"CopyrightDate": "",
"CopyrightName": "",
"LicenseName": "",
"LicenseDescription": "",
"Dependencies": [
"cpl-core>=2022.10.0.post2"
],
"DevDependencies": [
"cpl-cli==2022.10.0"
],
"PythonVersion": ">=3.10.4",
"PythonPath": {
"linux": ""
},
"Classifiers": []
},
"BuildSettings": {
"ProjectType": "library",
"SourcePath": "",
"OutputPath": "../../dist",
"Main": "permission.main",
"EntryPoint": "permission",
"IncludePackageData": false,
"Included": [],
"Excluded": [
"*/__pycache__",
"*/logs",
"*/tests"
],
"PackageData": {},
"ProjectReferences": []
}
}

28
kdb-bot/src/modules/permission/permission_module.py Normal file
View File

@@ -0,0 +1,28 @@
from cpl_core.configuration import ConfigurationABC
from cpl_core.dependency_injection import ServiceCollectionABC
from cpl_core.environment import ApplicationEnvironmentABC
from cpl_discord.discord_event_types_enum import DiscordEventTypesEnum
from cpl_discord.service.discord_collection_abc import DiscordCollectionABC
from bot_core.abc.module_abc import ModuleABC
from bot_core.configuration.feature_flags_enum import FeatureFlagsEnum
from modules.permission.abc.permission_service_abc import PermissionServiceABC
from modules.permission.events.permission_on_member_update_event import PermissionOnMemberUpdateEvent
from modules.permission.events.permission_on_ready_event import PermissionOnReadyEvent
from modules.permission.service.permission_service import PermissionService
class PermissionModule(ModuleABC):
def __init__(self, dc: DiscordCollectionABC):
ModuleABC.__init__(self, dc, FeatureFlagsEnum.permission_module)
def configure_configuration(self, config: ConfigurationABC, env: ApplicationEnvironmentABC):
pass
def configure_services(self, services: ServiceCollectionABC, env: ApplicationEnvironmentABC):
services.add_singleton(PermissionServiceABC, PermissionService)
# commands
# events
self._dc.add_event(DiscordEventTypesEnum.on_ready.value, PermissionOnReadyEvent)
self._dc.add_event(DiscordEventTypesEnum.on_member_update.value, PermissionOnMemberUpdateEvent)

26
kdb-bot/src/modules/permission/service/__init__.py Normal file
View File

@@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
"""
bot Keksdose bot
~~~~~~~~~~~~~~~~~~~
Discord bot for the Keksdose discord Server
:copyright: (c) 2022 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'modules.permission.service'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2022 sh-edraft.de'
__version__ = '0.2.3'
from collections import namedtuple
# imports
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='0', minor='2', micro='3')

135
kdb-bot/src/modules/permission/service/permission_service.py Normal file
View File

@@ -0,0 +1,135 @@
import discord
from cpl_core.logging import LoggerABC
from cpl_core.configuration import ConfigurationABC
from cpl_discord.service import DiscordBotServiceABC
from bot_core.configuration.bot_settings import BotSettings
from modules.permission.abc.permission_service_abc import PermissionServiceABC
from modules.permission.configuration.permission_server_settings import PermissionServerSettings
class PermissionService(PermissionServiceABC):
def __init__(
self,
logger: LoggerABC,
bot: DiscordBotServiceABC,
config: ConfigurationABC,
bot_settings: BotSettings
):
PermissionServiceABC.__init__(self)
self._logger = logger
self._bot = bot
self._config = config
self._admin_role_ids: dict[int, list[int]] = {}
self._admin_roles: dict[int, list[discord.Role]] = {}
self._admins: dict[int, list[discord.Member]] = {}
self._moderator_role_ids: dict[int, list[int]] = {}
self._moderator_roles: dict[int, list[discord.Role]] = {}
self._moderators: dict[int, list[discord.Member]] = {}
self._technician_ids: list[int] = bot_settings.technicians
self._technicians: list[discord.Member] = []
def on_ready(self):
for guild in self._bot.guilds:
guild: discord.Guild = guild
self._logger.debug(__name__, f'Validate permission settings')
for technician_id in self._technician_ids:
technician = guild.get_member(technician_id)
if technician is None:
continue
self._technicians.append(technician)
settings: PermissionServerSettings = self._config.get_configuration(f'PermissionServerSettings_{guild.id}')
if settings is None:
self._logger.error(__name__, 'Permission settings not found')
return
self._admin_role_ids[guild.id] = settings.admin_roles
self._moderator_role_ids[guild.id] = settings.moderator_roles
admin_roles = []
admins = []
mod_roles = []
mods = []
for role in guild.roles:
role: discord.Role = role
if role.id in self._admin_role_ids[guild.id]:
admin_roles.append(role)
self._logger.trace(__name__, f'Added admin role {role}')
for member in role.members:
admins.append(member)
self._logger.trace(__name__, f'Added admin {member}')
if role.id in self._moderator_role_ids[guild.id]:
mod_roles.append(role)
self._logger.trace(__name__, f'Added moderator role {role}')
for member in role.members:
mods.append(member)
self._logger.trace(__name__, f'Added moderator {member}')
self._admin_roles[guild.id] = admin_roles
self._admins[guild.id] = admins
self._moderator_roles[guild.id] = mod_roles
self._moderators[guild.id] = mods
def on_member_update(self, before: discord.Member, after: discord.Member):
g_id = after.guild.id
for admin_role in self._admin_roles[g_id]:
if admin_role in before.roles and admin_role not in after.roles:
self._admins[g_id].remove(after)
self._logger.trace(__name__, f'Removed {after.id} from admins')
elif admin_role in after.roles and admin_role not in before.roles:
self._admins[g_id].append(after)
self._logger.trace(__name__, f'Added {after.id} to admins')
for moderator_role in self._moderator_roles[g_id]:
if moderator_role in before.roles and moderator_role not in after.roles:
self._moderators[g_id].remove(after)
self._logger.trace(__name__, f'Removed {after.id} from moderators')
elif moderator_role in after.roles and moderator_role not in before.roles:
self._moderators[g_id].append(after)
self._logger.trace(__name__, f'Added {after.id} to moderators')
def get_admin_role_ids(self, g_id: int) -> list[int]:
return self._admin_role_ids[g_id]
def get_admin_roles(self, g_id: int) -> list[discord.Role]:
return self._admin_roles[g_id]
def get_admins(self, g_id: int) -> list[discord.Member]:
return self._admins[g_id]
def get_moderator_role_ids(self, g_id: int) -> list[int]:
return self._moderator_role_ids[g_id]
def get_moderator_roles(self, g_id: int) -> list[discord.Role]:
return self._moderator_roles[g_id]
def get_moderators(self, g_id: int) -> list[discord.Member]:
return self._moderators[g_id]
def get_technicians(self) -> list[discord.Member]:
return self._technicians
def is_member_admin(self, member: discord.Member) -> bool:
return member.guild.id in self._admins and member in self._admins[member.guild.id]
def is_member_moderator(self, member: discord.Member) -> bool:
return member.guild.id in self._moderators \
and member in self._moderators[member.guild.id] or self.is_member_admin(member)
def is_member_technician(self, member: discord.Member) -> bool:
return member in self._technicians