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

Moved folders #405

Browse Source
This commit is contained in:
Sven Heidemann
2023-10-13 17:10:00 +02:00
parent eb32bec43c
commit f435d3dd48
807 changed files with 3801 additions and 1297 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
bot/src/modules/level/__init__.py Normal file
View File

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

26
bot/src/modules/level/command/__init__.py Normal file
View File

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

676
bot/src/modules/level/command/level_group.py Normal file
View File

@@ -0,0 +1,676 @@
from typing import List as TList
import discord
from cpl_core.configuration import ConfigurationABC
from cpl_core.database.context import DatabaseContextABC
from cpl_discord.command import DiscordCommandABC
from cpl_discord.container import Guild, Role
from cpl_discord.service import DiscordBotServiceABC
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_abc import ClientUtilsABC
from bot_core.abc.message_service_abc import MessageServiceABC
from bot_core.configuration.feature_flags_enum import FeatureFlagsEnum
from bot_core.configuration.feature_flags_settings import FeatureFlagsSettings
from bot_core.helper.command_checks import CommandChecks
from bot_core.logging.command_logger import CommandLogger
from bot_data.abc.level_repository_abc import LevelRepositoryABC
from bot_data.abc.server_repository_abc import ServerRepositoryABC
from bot_data.abc.user_repository_abc import UserRepositoryABC
from bot_data.model.level import Level
from bot_data.model.server_config import ServerConfig
from modules.level.level_seeder import LevelSeeder
from modules.level.service.level_service import LevelService
from modules.permission.abc.permission_service_abc import PermissionServiceABC
class LevelGroup(DiscordCommandABC):
def __init__(
self,
config: ConfigurationABC,
logger: CommandLogger,
message_service: MessageServiceABC,
bot: DiscordBotServiceABC,
client_utils: ClientUtilsABC,
permission_service: PermissionServiceABC,
translate: TranslatePipe,
db: DatabaseContextABC,
levels: LevelRepositoryABC,
servers: ServerRepositoryABC,
users: UserRepositoryABC,
level_service: LevelService,
level_seeder: LevelSeeder,
):
DiscordCommandABC.__init__(self)
self._config = config
self._logger = logger
self._message_service = message_service
self._bot = bot
self._client_utils = client_utils
self._permissions = permission_service
self._t = translate
self._db = db
self._levels = levels
self._servers = servers
self._users = users
self._level_service = level_service
self._level_seeder = level_seeder
self._colors = [
("blue", discord.Colour.blue().to_rgb()),
("dark_blue", discord.Colour.dark_blue().to_rgb()),
("dark_gold", discord.Colour.dark_gold().to_rgb()),
("dark_gray", discord.Colour.dark_gray().to_rgb()),
("dark_green", discord.Colour.dark_green().to_rgb()),
("dark_grey", discord.Colour.dark_grey().to_rgb()),
("dark_magenta", discord.Colour.dark_magenta().to_rgb()),
("dark_orange", discord.Colour.dark_orange().to_rgb()),
("dark_purple", discord.Colour.dark_purple().to_rgb()),
("dark_red", discord.Colour.dark_red().to_rgb()),
("dark_teal", discord.Colour.dark_teal().to_rgb()),
("default", discord.Colour.default().to_rgb()),
("gold", discord.Colour.gold().to_rgb()),
("green", discord.Colour.green().to_rgb()),
("greyple", discord.Colour.greyple().to_rgb()),
("light_grey", discord.Colour.light_grey().to_rgb()),
("magenta", discord.Colour.magenta().to_rgb()),
("orange", discord.Colour.orange().to_rgb()),
("purple", discord.Colour.purple().to_rgb()),
("red", discord.Colour.red().to_rgb()),
("teal", discord.Colour.teal().to_rgb()),
("yellow", discord.Colour.yellow().to_rgb()),
]
self._logger.trace(__name__, f"Loaded command service: {type(self).__name__}")
async def _seed_levels(self, ctx: Context):
# send message to ctx.channel because send_ctx_msg resolves ctx
try:
start = await self._message_service.send_ctx_msg(
ctx,
self._t.transform("modules.level.seeding_started"),
is_persistent=True,
)
await self._level_seeder.seed()
end = await self._message_service.send_ctx_msg(
ctx,
self._t.transform("modules.level.seeding_finished"),
is_persistent=True,
)
await self._message_service.delete_message(start)
await self._message_service.delete_message(end)
except Exception as e:
self._logger.error(__name__, f"Level seeding failed", e)
await self._message_service.send_ctx_msg(
ctx, self._t.transform("modules.level.seeding_failed")
)
async def _level_auto_complete(
self, interaction: discord.Interaction, current: str
) -> TList[app_commands.Choice[str]]:
server = self._servers.get_server_by_discord_id(interaction.guild.id)
levels = self._levels.get_levels_by_server_id(server.id).select(
lambda l: l.name
)
return [
app_commands.Choice(name=level, value=level)
for level in self._client_utils.get_auto_complete_list(levels, current)
]
@commands.hybrid_group()
@commands.guild_only()
async def level(self, ctx: Context):
pass
@level.command(alias="levels")
@commands.guild_only()
@CommandChecks.check_is_ready()
@CommandChecks.check_is_member_moderator()
async def list(self, ctx: Context, wait: int = None):
self._logger.debug(__name__, f"Received command level list {ctx}")
if ctx.guild is None:
return
server_config: ServerConfig = self._config.get_configuration(
f"ServerConfig_{ctx.guild.id}"
)
if not FeatureFlagsSettings.get_flag_from_dict(
server_config.feature_flags, FeatureFlagsEnum.level_module
):
return
server = self._servers.get_server_by_discord_id(ctx.guild.id)
levels = self._levels.get_levels_by_server_id(server.id)
if levels.count() < 1:
await self._message_service.send_ctx_msg(
ctx, self._t.transform("modules.level.error.nothing_found")
)
self._logger.trace(__name__, f"Finished command level list")
return
level_name = ""
xp = ""
permissions = ""
for level in levels:
level_name += f"\n{level.name}"
xp += f"\n{level.min_xp}"
permissions += f"\n{level.permissions}"
embed = discord.Embed(
title=self._t.transform("modules.level.list.title"),
description=self._t.transform("modules.level.list.description"),
color=int("ef9d0d", 16),
)
embed.add_field(
name=self._t.transform("modules.level.list.name"),
value=level_name,
inline=True,
)
embed.add_field(
name=self._t.transform("modules.level.list.min_xp"), value=xp, inline=True
)
embed.add_field(
name=self._t.transform("modules.level.list.permission_int"),
value=permissions,
inline=True,
)
await self._message_service.send_ctx_msg(ctx, embed, wait_before_delete=wait)
self._logger.trace(__name__, f"Finished command level list")
@level.command()
@commands.guild_only()
@CommandChecks.check_is_ready()
@CommandChecks.check_is_member_admin()
async def create(
self, ctx: Context, name: str, color: str, min_xp: int, permissions: int
):
self._logger.debug(__name__, f"Received command level create {ctx}")
if ctx.guild is None:
return
server_config: ServerConfig = self._config.get_configuration(
f"ServerConfig_{ctx.guild.id}"
)
if not FeatureFlagsSettings.get_flag_from_dict(
server_config.feature_flags, FeatureFlagsEnum.level_module
):
return
try:
color = hex(discord.Colour.from_str(color).value)
except Exception as e:
self._logger.error(__name__, f"Error parsing color {color}", e)
return
try:
permissions = discord.Permissions(permissions).value
except Exception as e:
self._logger.error(__name__, f"Error parsing permissions {permissions}", e)
return
if ctx.guild is None:
return
server = self._servers.get_server_by_discord_id(ctx.guild.id)
level = Level(name, color, min_xp, permissions, server)
levels = self._levels.get_levels_by_server_id(server.id)
if levels.where(lambda l: l.name == level.name).first_or_default() is not None:
self._logger.debug(__name__, f"Level with name {level.name} already exists")
await self._message_service.send_ctx_msg(
ctx,
self._t.transform(
"modules.level.error.level_with_name_already_exists"
).format(level.name),
)
elif (
levels.where(lambda l: l.min_xp == level.min_xp).first_or_default()
is not None
):
self._logger.debug(
__name__,
f"Level with min_xp {level.min_xp} already exists {level.name}",
)
found_level = levels.where(
lambda l: l.min_xp == level.min_xp
).first_or_default()
await self._message_service.send_ctx_msg(
ctx,
self._t.transform(
"modules.level.error.level_with_xp_already_exists"
).format(found_level.name, found_level.min_xp),
)
else:
try:
self._levels.add_level(level)
self._db.save_changes()
self._logger.info(
__name__,
f"Saved level {name} with color {color}, min_xp {min_xp} and permissions {permissions}",
)
except Exception as e:
self._logger.error(
__name__,
f"Could not save level {name} with color {color}, min_xp {min_xp} and permissions {permissions}",
e,
)
else:
await self._message_service.send_ctx_msg(
ctx,
self._t.transform("modules.level.create.created").format(
name, permissions
),
)
await self._seed_levels(ctx)
self._logger.trace(__name__, f"Finished command level create")
@create.autocomplete("color")
async def create_color_autocomplete(
self, interaction: discord.Interaction, current: str
) -> TList[app_commands.Choice[str]]:
# value in rg format see:
# https://discordpy.readthedocs.io/en/latest/api.html#discord.Colour.to_rgb
return [
app_commands.Choice(
name=self._t.transform(f"common.colors.{color}"),
value=f"rgb({code[0]}, {code[1]}, {code[2]})",
)
for color, code in self._colors
]
@level.command()
@commands.guild_only()
@CommandChecks.check_is_ready()
@CommandChecks.check_is_member_admin()
async def edit(
self,
ctx: Context,
level: str,
name: str = None,
color: str = None,
min_xp: int = None,
permissions: int = None,
):
self._logger.debug(__name__, f"Received command level edit {ctx}")
if ctx.guild is None:
return
server_config: ServerConfig = self._config.get_configuration(
f"ServerConfig_{ctx.guild.id}"
)
if not FeatureFlagsSettings.get_flag_from_dict(
server_config.feature_flags, FeatureFlagsEnum.level_module
):
return
server = self._servers.get_server_by_discord_id(ctx.guild.id)
level_from_db = (
self._levels.get_levels_by_server_id(server.id)
.where(lambda l: l.name == level)
.single_or_default()
)
if level_from_db is None:
await self._message_service.send_ctx_msg(
ctx, self._t.transform("modules.level.edit.not_found").format(level)
)
return
guild: Guild = self._bot.guilds.where(lambda g: g == ctx.guild).single()
role: Role = guild.roles.where(lambda r: r.name == level_from_db.name).single()
if name is not None:
level_from_db.name = name
if color is not None:
try:
level_from_db.color = hex(discord.Colour.from_str(color).value)
except Exception as e:
await self._message_service.send_ctx_msg(
ctx,
self._t.transform("modules.level.edit.color_invalid").format(color),
)
self._logger.error(__name__, f"Error parsing color {color}", e)
return
if min_xp is not None:
level_from_db.min_xp = min_xp
if permissions is not None:
try:
level_from_db.permissions = discord.Permissions(permissions).value
except Exception as e:
await self._message_service.send_ctx_msg(
ctx,
self._t.transform("modules.level.edit.permission_invalid").format(
permissions
),
)
self._logger.error(
__name__, f"Error parsing permissions {permissions}", e
)
return
try:
self._levels.update_level(level_from_db)
self._db.save_changes()
await role.edit(
name=level_from_db.name,
permissions=discord.Permissions(level_from_db.permissions),
colour=discord.Colour(int(level_from_db.color, 16)),
)
self._logger.info(
__name__,
f"Saved level {level_from_db.name} with color {level_from_db.color}, min_xp {level_from_db.min_xp} and permissions {level_from_db.permissions}",
)
except Exception as e:
self._logger.error(
__name__,
f"Could not save level {level} with color {color}, min_xp {min_xp} and permissions {permissions}",
e,
)
else:
await self._message_service.send_ctx_msg(
ctx, self._t.transform("modules.level.edit.edited").format(level)
)
await self._seed_levels(ctx)
self._logger.trace(__name__, f"Finished command level edit")
@edit.autocomplete("level")
async def edit_autocomplete(
self, interaction: discord.Interaction, current: str
) -> TList[app_commands.Choice[str]]:
return await self._level_auto_complete(interaction, current)
@edit.autocomplete("color")
async def edit_color_autocomplete(
self, interaction: discord.Interaction, current: str
) -> TList[app_commands.Choice[str]]:
# value in rg format see:
# https://discordpy.readthedocs.io/en/latest/api.html#discord.Colour.to_rgb
return [
app_commands.Choice(
name=self._t.transform(f"common.colors.{color}"),
value=f"rgb({code[0]}, {code[1]}, {code[2]})",
)
for color, code in self._colors
]
@level.command()
@commands.guild_only()
@CommandChecks.check_is_ready()
@CommandChecks.check_is_member_admin()
async def remove(self, ctx: Context, level: str):
self._logger.debug(__name__, f"Received command level remove {ctx}")
if ctx.guild is None:
return
server_config: ServerConfig = self._config.get_configuration(
f"ServerConfig_{ctx.guild.id}"
)
if not FeatureFlagsSettings.get_flag_from_dict(
server_config.feature_flags, FeatureFlagsEnum.level_module
):
return
server = self._servers.get_server_by_discord_id(ctx.guild.id)
level_from_db = (
self._levels.get_levels_by_server_id(server.id)
.where(lambda l: l.name == level)
.first_or_default()
)
if level_from_db is None:
self._logger.debug(__name__, f"level {level} not found")
await self._message_service.send_ctx_msg(
ctx,
self._t.transform("modules.level.remove.error.not_found").format(level),
)
self._logger.trace(__name__, f"Finished command level remove")
return
try:
self._levels.delete_level(level_from_db)
self._db.save_changes()
guild: Guild = self._bot.guilds.where(lambda g: g == ctx.guild).single()
role: Role = guild.roles.where(
lambda r: r.name == level
).single_or_default()
if role is not None:
await role.delete()
self._logger.info(__name__, f"Removed level {level}")
except Exception as e:
self._logger.error(__name__, f"Could not remove level {level}", e)
else:
await self._message_service.send_ctx_msg(
ctx, self._t.transform("modules.level.remove.success").format(level)
)
await self._seed_levels(ctx)
self._logger.trace(__name__, f"Finished command level remove")
@remove.autocomplete("level")
async def remove_autocomplete(
self, interaction: discord.Interaction, current: str
) -> TList[app_commands.Choice[str]]:
return await self._level_auto_complete(interaction, current)
@level.command()
@commands.guild_only()
@CommandChecks.check_is_ready()
@CommandChecks.check_is_member_moderator()
async def down(self, ctx: Context, member: discord.Member):
self._logger.debug(__name__, f"Received command level down {ctx} {member}")
if ctx.guild is None:
return
server_config: ServerConfig = self._config.get_configuration(
f"ServerConfig_{ctx.guild.id}"
)
if not FeatureFlagsSettings.get_flag_from_dict(
server_config.feature_flags, FeatureFlagsEnum.level_module
):
return
if member.bot:
return
server = self._servers.get_server_by_discord_id(ctx.guild.id)
user = self._users.get_user_by_discord_id_and_server_id(member.id, server.id)
level = self._level_service.get_level(user)
levels = self._levels.get_levels_by_server_id(server.id).order_by(
lambda l: l.min_xp
)
if level == levels.first():
await self._message_service.send_ctx_msg(
ctx,
self._t.transform("modules.level.down.already_first").format(
member.mention
),
)
self._logger.trace(__name__, f"Finished command level down")
return
try:
new_level = levels.where(lambda l: l.min_xp < level.min_xp).last()
user.xp = new_level.min_xp
self._users.update_user(user)
self._db.save_changes()
await self._message_service.send_ctx_msg(
ctx,
self._t.transform("modules.level.down.success").format(
member.mention, new_level.name
),
)
await self._level_service.set_level(user)
except Exception as e:
self._logger.error(
__name__, f"Cannot level down {member.name} with level {level.name}", e
)
await self._message_service.send_ctx_msg(
ctx,
self._t.transform("modules.level.down.failed").format(member.mention),
)
self._logger.trace(__name__, f"Finished command level down")
@level.command()
@commands.guild_only()
@CommandChecks.check_is_ready()
@CommandChecks.check_is_member_moderator()
async def up(self, ctx: Context, member: discord.Member):
self._logger.debug(__name__, f"Received command level up {ctx} {member}")
if ctx.guild is None:
return
server_config: ServerConfig = self._config.get_configuration(
f"ServerConfig_{ctx.guild.id}"
)
if not FeatureFlagsSettings.get_flag_from_dict(
server_config.feature_flags, FeatureFlagsEnum.level_module
):
return
if member.bot:
return
server = self._servers.get_server_by_discord_id(ctx.guild.id)
user = self._users.get_user_by_discord_id_and_server_id(member.id, server.id)
level = self._level_service.get_level(user)
levels = self._levels.get_levels_by_server_id(server.id).order_by(
lambda l: l.min_xp
)
if level.name == levels.last().name:
await self._message_service.send_ctx_msg(
ctx,
self._t.transform("modules.level.up.already_last").format(
member.mention
),
)
self._logger.trace(__name__, f"Finished command level up")
return
try:
new_level = levels.where(lambda l: l.min_xp > level.min_xp).first()
user.xp = new_level.min_xp
self._users.update_user(user)
self._db.save_changes()
await self._message_service.send_ctx_msg(
ctx,
self._t.transform("modules.level.up.success").format(
member.mention, new_level.name
),
)
await self._level_service.set_level(user)
except Exception as e:
self._logger.error(
__name__, f"Cannot level up {member.name} with level {level.name}", e
)
await self._message_service.send_ctx_msg(
ctx, self._t.transform("modules.level.up.failed").format(member.mention)
)
self._logger.trace(__name__, f"Finished command level up")
@level.command()
@commands.guild_only()
@CommandChecks.check_is_ready()
@CommandChecks.check_is_member_moderator()
async def set(self, ctx: Context, member: discord.Member, level: str):
self._logger.debug(__name__, f"Received command level up {ctx} {member}")
if ctx.guild is None:
return
server_config: ServerConfig = self._config.get_configuration(
f"ServerConfig_{ctx.guild.id}"
)
if not FeatureFlagsSettings.get_flag_from_dict(
server_config.feature_flags, FeatureFlagsEnum.level_module
):
return
if member.bot:
return
server = self._servers.get_server_by_discord_id(ctx.guild.id)
user = self._users.get_user_by_discord_id_and_server_id(member.id, server.id)
current_level = self._level_service.get_level(user)
new_level = (
self._levels.get_levels_by_server_id(server.id)
.where(lambda l: l.name == level)
.single_or_default()
)
if new_level is None:
await self._message_service.send_ctx_msg(
ctx, self._t.transform("modules.level.set.not_found").format(level)
)
self._logger.trace(__name__, f"Finished command level set")
return
if current_level.name == level:
await self._message_service.send_ctx_msg(
ctx,
self._t.transform("modules.level.set.already_level").format(
member.mention, level
),
)
self._logger.trace(__name__, f"Finished command level set")
return
try:
user.xp = new_level.min_xp
self._users.update_user(user)
self._db.save_changes()
await self._message_service.send_ctx_msg(
ctx,
self._t.transform("modules.level.set.success").format(
member.mention, new_level.name
),
)
await self._level_service.set_level(user)
except Exception as e:
self._logger.error(
__name__, f"Cannot set level {level} for {member.name}", e
)
await self._message_service.send_ctx_msg(
ctx,
self._t.transform("modules.level.set.failed").format(member.mention),
)
self._logger.trace(__name__, f"Finished command level set")
@set.autocomplete("level")
async def set_autocomplete(
self, interaction: discord.Interaction, current: str
) -> TList[app_commands.Choice[str]]:
return await self._level_auto_complete(interaction, current)
@level.command()
@commands.guild_only()
@CommandChecks.check_is_ready()
@CommandChecks.check_is_member_moderator()
async def reload(self, ctx: Context):
self._logger.debug(__name__, f"Received command level reload {ctx}")
if ctx.guild is None:
return
server_config: ServerConfig = self._config.get_configuration(
f"ServerConfig_{ctx.guild.id}"
)
if not FeatureFlagsSettings.get_flag_from_dict(
server_config.feature_flags, FeatureFlagsEnum.level_module
):
return
await self._seed_levels(ctx)
self._logger.trace(__name__, f"Finished command level reload")

26
bot/src/modules/level/configuration/__init__.py Normal file
View File

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

33
bot/src/modules/level/configuration/default_level_settings.py Normal file
View File

@@ -0,0 +1,33 @@
from cpl_core.configuration.configuration_model_abc import ConfigurationModelABC
from cpl_query.extension import List
from bot_data.model.level import Level
class DefaultLevelSettings(ConfigurationModelABC):
def __init__(self, level_header: str = None, levels: list = None):
ConfigurationModelABC.__init__(self)
self._levels = List(Level)
self._level_header = level_header
if levels is None:
return
for level in levels:
self._levels.append(
Level(
level["Name"],
level["Color"],
int(level["MinXp"]),
int(level["Permissions"]),
None,
)
)
@property
def levels(self) -> List[Level]:
return self._levels
@property
def level_header(self) -> str:
return self._level_header

31
bot/src/modules/level/default-level.json Normal file
View File

@@ -0,0 +1,31 @@
{
"DefaultLevel": {
"LevelHeader": "~~~ Level ~~~",
"Levels": [
{
"Name": "Newbie",
"Color": "0x1abc9c",
"MinXp": 0,
"Permissions": 968552209984
},
{
"Name": "Keks",
"Color": "0x2ecc71",
"MinXp": 100,
"Permissions": 1002928856640
},
{
"Name": "Doppelkeks",
"Color": "0x3498db",
"MinXp": 200,
"Permissions": 1071849660224
},
{
"Name": "Auror",
"Color": "0xf1c40f",
"MinXp": 300,
"Permissions": 1089042120513
}
]
}
}

26
bot/src/modules/level/events/__init__.py Normal file
View File

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

33
bot/src/modules/level/events/level_on_member_join_event.py Normal file
View File

@@ -0,0 +1,33 @@
import discord
from cpl_core.configuration import ConfigurationABC
from cpl_discord.events import OnMemberJoinABC
from bot_core.configuration.feature_flags_enum import FeatureFlagsEnum
from bot_core.configuration.feature_flags_settings import FeatureFlagsSettings
from bot_core.helper.event_checks import EventChecks
from bot_core.logging.message_logger import MessageLogger
from bot_data.model.server_config import ServerConfig
from modules.level.service.level_service import LevelService
class LevelOnMemberJoinEvent(OnMemberJoinABC):
def __init__(
self, config: ConfigurationABC, logger: MessageLogger, level: LevelService
):
OnMemberJoinABC.__init__(self)
self._config = config
self._logger = logger
self._level = level
@EventChecks.check_is_ready()
async def on_member_join(self, member: discord.Member):
self._logger.debug(__name__, f"Module {type(self)} started")
server_config: ServerConfig = self._config.get_configuration(
f"ServerConfig_{member.guild.id}"
)
if not FeatureFlagsSettings.get_flag_from_dict(
server_config.feature_flags, FeatureFlagsEnum.level_module
):
return
await self._level.check_level(member)

42
bot/src/modules/level/events/level_on_message_event.py Normal file
View File

@@ -0,0 +1,42 @@
import discord
from cpl_core.configuration import ConfigurationABC
from cpl_discord.events import OnMessageABC
from bot_core.configuration.feature_flags_enum import FeatureFlagsEnum
from bot_core.configuration.feature_flags_settings import FeatureFlagsSettings
from bot_core.helper.event_checks import EventChecks
from bot_core.logging.message_logger import MessageLogger
from bot_data.model.server_config import ServerConfig
from modules.level.service.level_service import LevelService
class LevelOnMessageEvent(OnMessageABC):
def __init__(
self, config: ConfigurationABC, logger: MessageLogger, level: LevelService
):
OnMessageABC.__init__(self)
self._config = config
self._logger = logger
self._level = level
@EventChecks.check_is_ready()
async def on_message(self, message: discord.Message):
self._logger.debug(__name__, f"Module {type(self)} started")
if message.guild is None:
return
if message.author.bot:
return
server_config: ServerConfig = self._config.get_configuration(
f"ServerConfig_{message.guild.id}"
)
if not FeatureFlagsSettings.get_flag_from_dict(
server_config.feature_flags, FeatureFlagsEnum.level_module
):
return
try:
await self._level.check_level(message.author)
except Exception as e:
self._logger.error(__name__, f"Level check by message failed", e)

48
bot/src/modules/level/events/level_on_raw_reaction_add_event.py Normal file
View File

@@ -0,0 +1,48 @@
from cpl_core.configuration import ConfigurationABC
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_core.configuration.feature_flags_enum import FeatureFlagsEnum
from bot_core.configuration.feature_flags_settings import FeatureFlagsSettings
from bot_core.helper.event_checks import EventChecks
from bot_data.model.server_config import ServerConfig
from modules.level.service.level_service import LevelService
class LevelOnRawReactionAddEvent(OnRawReactionAddABC):
def __init__(
self,
config: ConfigurationABC,
logger: LoggerABC,
bot: DiscordBotServiceABC,
level: LevelService,
):
OnRawReactionAddABC.__init__(self)
self._config = config
self._logger = logger
self._bot = bot
self._level = level
@EventChecks.check_is_ready()
async def on_raw_reaction_add(self, payload: RawReactionActionEvent):
self._logger.debug(__name__, f"Module {type(self)} started")
server_config: ServerConfig = self._config.get_configuration(
f"ServerConfig_{payload.guild_id}"
)
if not FeatureFlagsSettings.get_flag_from_dict(
server_config.feature_flags, FeatureFlagsEnum.level_module
):
return
try:
self._logger.trace(__name__, f"Handle reaction {payload} for level")
guild = self._bot.get_guild(payload.guild_id)
member = guild.get_member(payload.user_id)
await self._level.check_level(member)
except Exception as e:
self._logger.error(__name__, f"Level check by message failed", e)

48
bot/src/modules/level/events/level_on_raw_reaction_remove_event.py Normal file
View File

@@ -0,0 +1,48 @@
from cpl_core.configuration import ConfigurationABC
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_core.configuration.feature_flags_enum import FeatureFlagsEnum
from bot_core.configuration.feature_flags_settings import FeatureFlagsSettings
from bot_core.helper.event_checks import EventChecks
from bot_data.model.server_config import ServerConfig
from modules.level.service.level_service import LevelService
class LevelOnRawReactionRemoveEvent(OnRawReactionRemoveABC):
def __init__(
self,
config: ConfigurationABC,
logger: LoggerABC,
bot: DiscordBotServiceABC,
level: LevelService,
):
OnRawReactionRemoveABC.__init__(self)
self._config = config
self._logger = logger
self._bot = bot
self._level = level
@EventChecks.check_is_ready()
async def on_raw_reaction_remove(self, payload: RawReactionActionEvent):
self._logger.debug(__name__, f"Module {type(self)} started")
server_config: ServerConfig = self._config.get_configuration(
f"ServerConfig_{payload.guild_id}"
)
if not FeatureFlagsSettings.get_flag_from_dict(
server_config.feature_flags, FeatureFlagsEnum.level_module
):
return
try:
self._logger.trace(__name__, f"Handle reaction {payload} for level")
guild = self._bot.get_guild(payload.guild_id)
member = guild.get_member(payload.user_id)
await self._level.check_level(member)
except Exception as e:
self._logger.error(__name__, f"Level check by message failed", e)

40
bot/src/modules/level/events/level_on_voice_state_update_event.py Normal file
View File

@@ -0,0 +1,40 @@
import discord
from cpl_core.configuration import ConfigurationABC
from cpl_core.logging import LoggerABC
from cpl_discord.events import OnVoiceStateUpdateABC
from bot_core.configuration.feature_flags_enum import FeatureFlagsEnum
from bot_core.configuration.feature_flags_settings import FeatureFlagsSettings
from bot_core.helper.event_checks import EventChecks
from bot_data.model.server_config import ServerConfig
from modules.level.service.level_service import LevelService
class LevelOnVoiceStateUpdateEvent(OnVoiceStateUpdateABC):
def __init__(
self, config: ConfigurationABC, logger: LoggerABC, level: LevelService
):
OnVoiceStateUpdateABC.__init__(self)
self._config = config
self._logger = logger
self._level = level
self._logger.info(__name__, f"Module {type(self)} loaded")
@EventChecks.check_is_ready()
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")
server_config: ServerConfig = self._config.get_configuration(
f"ServerConfig_{member.guild.id}"
)
if not FeatureFlagsSettings.get_flag_from_dict(
server_config.feature_flags, FeatureFlagsEnum.level_module
):
return
await self._level.check_level(member)

44
bot/src/modules/level/level.json Normal file
View File

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

66
bot/src/modules/level/level_module.py Normal file
View File

@@ -0,0 +1,66 @@
import os
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 bot_data.abc.data_seeder_abc import DataSeederABC
from modules.level.command.level_group import LevelGroup
from modules.level.events.level_on_member_join_event import LevelOnMemberJoinEvent
from modules.level.events.level_on_message_event import LevelOnMessageEvent
from modules.level.events.level_on_raw_reaction_add_event import (
LevelOnRawReactionAddEvent,
)
from modules.level.events.level_on_raw_reaction_remove_event import (
LevelOnRawReactionRemoveEvent,
)
from modules.level.events.level_on_voice_state_update_event import (
LevelOnVoiceStateUpdateEvent,
)
from modules.level.level_seeder import LevelSeeder
from modules.level.service.level_service import LevelService
class LevelModule(ModuleABC):
def __init__(self, dc: DiscordCollectionABC):
ModuleABC.__init__(self, dc, FeatureFlagsEnum.level_module)
def configure_configuration(
self, config: ConfigurationABC, env: ApplicationEnvironmentABC
):
cwd = env.working_directory
env.set_working_directory(os.path.dirname(os.path.realpath(__file__)))
config.add_json_file(f"default-level.json", optional=False)
env.set_working_directory(cwd)
def configure_services(
self, services: ServiceCollectionABC, env: ApplicationEnvironmentABC
):
services.add_transient(DataSeederABC, LevelSeeder)
services.add_transient(LevelService)
# commands
services.add_transient(LevelGroup)
# events
services.add_transient(
DiscordEventTypesEnum.on_message.value, LevelOnMessageEvent
)
services.add_transient(
DiscordEventTypesEnum.on_voice_state_update.value,
LevelOnVoiceStateUpdateEvent,
)
services.add_transient(
DiscordEventTypesEnum.on_member_join.value, LevelOnMemberJoinEvent
)
services.add_transient(
DiscordEventTypesEnum.on_raw_reaction_add.value, LevelOnRawReactionAddEvent
)
services.add_transient(
DiscordEventTypesEnum.on_raw_reaction_remove.value,
LevelOnRawReactionRemoveEvent,
)

152
bot/src/modules/level/level_seeder.py Normal file
View File

@@ -0,0 +1,152 @@
import discord
from cpl_core.configuration import ConfigurationABC
from cpl_core.database.context import DatabaseContextABC
from cpl_discord.container import Guild
from cpl_discord.service import DiscordBotServiceABC
from discord import Permissions, Colour
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.abc.data_seeder_abc import DataSeederABC
from bot_data.abc.server_repository_abc import ServerRepositoryABC
from bot_data.model.level import Level
from bot_data.model.server import Server
from bot_data.model.server_config import ServerConfig
from bot_data.service.level_repository_service import LevelRepositoryService
from modules.level.configuration.default_level_settings import DefaultLevelSettings
from modules.level.service.level_service import LevelService
class LevelSeeder(DataSeederABC):
def __init__(
self,
config: ConfigurationABC,
logger: DatabaseLogger,
levels: DefaultLevelSettings,
level_repo: LevelRepositoryService,
servers: ServerRepositoryABC,
level: LevelService,
db: DatabaseContextABC,
bot: DiscordBotServiceABC,
):
DataSeederABC.__init__(self)
self._config = config
self._logger = logger
self._levels = level_repo
self._servers = servers
self._level = level
self._db = db
self._bot = bot
self._level_header = levels.level_header
self._default_levels = levels.levels.order_by_descending(lambda l: l.min_xp)
async def _create_level(self, level: Level, guild: Guild, server: Server):
level.server = server
try:
if (
guild.roles.where(lambda r: r.name == level.name).first_or_default()
is None
):
await guild.create_role(
name=level.name,
colour=Colour(int(level.color, 16)),
hoist=False,
mentionable=True,
permissions=Permissions(level.permissions),
)
self._logger.debug(__name__, f"Created role {level.name}")
levels = self._levels.find_levels_by_server_id(server.id)
if (
levels is None
or levels.where(lambda l: l.name == level.name).first_or_default()
is None
):
self._levels.add_level(level)
self._logger.debug(__name__, f"Saved level {level.name}")
self._db.save_changes()
except discord.errors.Forbidden as e:
self._logger.error(__name__, f"Creating level failed", e)
level.permissions = 0
self._levels.update_level(level)
except Exception as e:
self._logger.error(__name__, f"Creating level failed", e)
async def seed(self):
# create levels
for guild in self._bot.guilds:
server_config: ServerConfig = self._config.get_configuration(
f"ServerConfig_{guild.id}"
)
if not FeatureFlagsSettings.get_flag_from_dict(
server_config.feature_flags, FeatureFlagsEnum.level_module
):
continue
created_default = False
if (
guild.roles.where(
lambda r: r.name == self._level_header
).first_or_default()
is None
):
await guild.create_role(name=self._level_header)
server = self._servers.find_server_by_discord_id(guild.id)
if server is None:
continue
levels = self._levels.find_levels_by_server_id(server.id)
if levels is not None and levels.count() > 0:
# create levels from db
for level in levels:
await self._create_level(level, guild, server)
self._logger.debug(__name__, f"Seeded levels")
else:
# create default levels
for level in self._default_levels:
created_default = True
await self._create_level(level, guild, server)
self._logger.debug(__name__, f"Seeded default levels")
if created_default:
continue
levels = levels.order_by_descending(lambda l: l.min_xp)
position_above_levels = (
guild.roles.where(lambda r: r.name == self._level_header)
.single()
.position
)
for role in guild.roles.order_by_descending(lambda r: r.position):
if levels.where(lambda l: l.name == role.name).count() == 0:
continue
new_position = position_above_levels - (
levels.index_of(
levels.where(lambda l: l.name == role.name).single()
)
+ 1
)
if new_position <= 0:
new_position = 1
try:
self._logger.debug(
__name__,
f"Moved {role.name} from {role.position} to {new_position}",
)
await role.edit(position=new_position)
except Exception as e:
self._logger.error(
__name__, f"Cannot change position of {role.name}", e
)
for m in guild.members:
await self._level.check_level(m)
self._logger.debug(__name__, f"Checked role order")

26
bot/src/modules/level/service/__init__.py Normal file
View File

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

120
bot/src/modules/level/service/level_service.py Normal file
View File

@@ -0,0 +1,120 @@
import discord
from cpl_core.configuration import ConfigurationABC
from cpl_core.database.context import DatabaseContextABC
from cpl_core.logging import LoggerABC
from cpl_discord.container import Guild, Role, Member
from cpl_discord.service import DiscordBotServiceABC
from cpl_translation import TranslatePipe
from bot_core.service.message_service import MessageService
from bot_data.model.level import Level
from bot_data.model.server_config import ServerConfig
from bot_data.model.user import User
from bot_data.service.level_repository_service import LevelRepositoryService
from bot_data.service.server_repository_service import ServerRepositoryService
from bot_data.service.user_repository_service import UserRepositoryService
class LevelService:
def __init__(
self,
config: ConfigurationABC,
logger: LoggerABC,
db: DatabaseContextABC,
levels: LevelRepositoryService,
users: UserRepositoryService,
servers: ServerRepositoryService,
bot: DiscordBotServiceABC,
message_service: MessageService,
t: TranslatePipe,
):
self._config = config
self._logger = logger
self._db = db
self._levels = levels
self._users = users
self._servers = servers
self._bot = bot
self._message_service = message_service
self._t = t
def get_level(self, user: User) -> Level:
levels_by_server = self._levels.get_levels_by_server_id(user.server.id)
if user.xp < 0:
return levels_by_server.order_by(lambda l: l.min_xp).first()
levels = levels_by_server.order_by(lambda l: l.min_xp).where(
lambda l: user.xp >= l.min_xp
)
if levels.count() == 0:
return levels_by_server.order_by(lambda l: l.min_xp).last()
return levels.last()
async def set_level(self, user: User):
level_names = self._levels.get_levels_by_server_id(user.server.id).select(
lambda l: l.name
)
guild: Guild = self._bot.guilds.where(
lambda g: g.id == user.server.discord_id
).single()
member: Member = guild.members.where(lambda m: m.id == user.discord_id).single()
level = self.get_level(user)
level_role: Role = guild.roles.where(lambda r: r.name == level.name).single()
if level_role in member.roles:
return
notification_needed = False
for role in member.roles.where(lambda r: r.name in level_names.to_list()):
try:
self._logger.debug(
__name__, f"Try to remove role {role.name} from {member.name}"
)
await member.remove_roles(role)
notification_needed = True
self._logger.info(
__name__, f"Removed role {role.name} from {member.name}"
)
except Exception as e:
self._logger.error(
__name__, f"Removing role {role.name} from {member.name} failed!", e
)
try:
self._logger.debug(
__name__, f"Try to add role {level_role.name} to {member.name}"
)
await member.add_roles(level_role)
self._logger.info(__name__, f"Add role {level_role.name} to {member.name}")
except Exception as e:
self._logger.error(
__name__, f"Adding role {level_role.name} to {member.name} failed!", e
)
if notification_needed:
settings: ServerConfig = self._config.get_configuration(
f"ServerConfig_{guild.id}"
)
await self._message_service.send_channel_message(
self._bot.get_channel(settings.notification_chat_id),
self._t.transform("modules.level.new_level_message").format(
member.mention, level.name
),
is_persistent=True,
)
async def check_level(self, member: discord.Member):
if member.bot:
return
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.id)
if user is None:
self._logger.warn(
__name__, f"User not found {member.guild.name}@{member.name}"
)
return
await self.set_level(user)