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

Added warning commands #35

Browse Source
This commit is contained in:
Sven Heidemann 2023-02-22 08:44:08 +01:00
parent 1eb625fe7a
commit 2674af64e9
11 changed files with 208 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
kdb-bot/src/bot/config

@ -1 +1 @@
Subproject commit 7986144705052ff38472a5d3f0776cb6c1752a55
Subproject commit 6e2ec8f2f88cca5355624da9c83c034949d12ae3

19
kdb-bot/src/bot/translation/de.json
View File

@ -209,12 +209,25 @@
"success": "Verlinkung wurde entfernt :D"
},
"warnings": {
"message": "Du wurdest verwarnt. Der Grund ist: {}",
"messages": {
"warned": "Du wurdest verwarnt. Der Grund ist: {}",
"team_warned": "{} wurde verwarnt. Der Grund ist: {}",
"removed": "Die Verwarnung '{}' wurde entfernt.",
"team_removed": "Die Verwarnung '{}' an {} wurde entfernt.",
"first": "Bei der nächsten verwarnung, wirst du auf das vorherige Level zurückgesetzt!",
"second": "Bei der nächsten verwarnung, wirst du auf das erste Level zurückgesetzt!",
"third": "Bei der nächsten verwarnung, wirst du gekickt und zurückgesetzt!",
"kick": "Ich musste {} aufgrund zu vieler Verwarnungen kicken"
"kick": "Ich musste {} aufgrund zu vieler Verwarnungen kicken",
"show": {
"id": "Id",
"description": "Beschreibung"
},
"add": {
"success": "Verwarnung wurde hinzugefügt :)",
"failed": "Verwarnung konnte nicht hinzugefügt werden :("
},
"remove": {
"success": "Verwarnung wurde entfernt :)",
"failed": "Verwarnung konnte nicht entfernt werden :("
}
}
},

0
kdb-bot/src/bot_data/abc/user_warning_repository_abc.py → kdb-bot/src/bot_data/abc/user_warnings_repository_abc.py
View File

2
kdb-bot/src/bot_data/data_module.py
View File

@ -21,7 +21,7 @@ from bot_data.abc.user_message_count_per_hour_repository_abc import (
UserMessageCountPerHourRepositoryABC,
)
from bot_data.abc.user_repository_abc import UserRepositoryABC
from bot_data.abc.user_warning_repository_abc import UserWarningsRepositoryABC
from bot_data.abc.user_warnings_repository_abc import UserWarningsRepositoryABC
from bot_data.service.api_key_repository_service import ApiKeyRepositoryService
from bot_data.service.auth_user_repository_service import AuthUserRepositoryService
from bot_data.service.auto_role_repository_service import AutoRoleRepositoryService

2
kdb-bot/src/bot_data/migration/user_warning_migration.py
View File

@ -21,10 +21,12 @@ class UserWarningMigration(MigrationABC):
CREATE TABLE IF NOT EXISTS `UserWarnings` (
`Id` BIGINT NOT NULL AUTO_INCREMENT,
`Description` VARCHAR(255) NOT NULL,
`UserId` BIGINT NOT NULL,
`Author` BIGINT NULL,
`CreatedAt` DATETIME(6),
`LastModifiedAt` DATETIME(6),
PRIMARY KEY(`Id`),
FOREIGN KEY (`UserId`) REFERENCES `Users`(`UserId`),
FOREIGN KEY (`Author`) REFERENCES `Users`(`UserId`)
);
"""

8
kdb-bot/src/bot_data/model/user_warnings.py
View File

@ -11,7 +11,7 @@ class UserWarnings(TableABC):
def __init__(
self,
description: str,
user: Optional[User],
user: User,
author: Optional[User],
created_at: datetime = None,
modified_at: datetime = None,
@ -34,6 +34,10 @@ class UserWarnings(TableABC):
def description(self) -> str:
return self._description
@property
def user(self) -> User:
return self._user
@property
def author(self) -> Optional[User]:
return self._author
@ -51,7 +55,7 @@ class UserWarnings(TableABC):
return str(
f"""
SELECT * FROM `UserWarnings`
WHERE `UserWarnings` = {id};
WHERE `Id` = {id};
"""
)

17
kdb-bot/src/bot_data/service/user_warnings_repository_service.py
View File

@ -4,8 +4,9 @@ from cpl_core.database.context import DatabaseContextABC
from cpl_query.extension import List
from bot_core.logging.database_logger import DatabaseLogger
from bot_data.abc.server_repository_abc import ServerRepositoryABC
from bot_data.abc.user_repository_abc import UserRepositoryABC
from bot_data.abc.user_warning_repository_abc import UserWarningsRepositoryABC
from bot_data.abc.user_warnings_repository_abc import UserWarningsRepositoryABC
from bot_data.model.user_warnings import UserWarnings
@ -15,6 +16,7 @@ class UserWarningsRepositoryService(UserWarningsRepositoryABC):
logger: DatabaseLogger,
db_context: DatabaseContextABC,
users: UserRepositoryABC,
servers: ServerRepositoryABC,
):
self._logger = logger
self._context = db_context
@ -30,12 +32,19 @@ class UserWarningsRepositoryService(UserWarningsRepositoryABC):
return value
def _from_result(self, sql_result: tuple) -> UserWarnings:
user = self._users.get_user_by_id(self._get_value_from_result(sql_result[2]))
author = None
author_id = self._get_value_from_result(sql_result[2])
if author_id is not None:
author = self._users.get_user_by_id(author_id)
return UserWarnings(
self._get_value_from_result(sql_result[1]),
self._get_value_from_result(sql_result[2]),
self._get_value_from_result(sql_result[3]),
user,
author,
self._get_value_from_result(sql_result[4]),
self._get_value_from_result(sql_result[0]),
self._get_value_from_result(sql_result[5]),
id=self._get_value_from_result(sql_result[0]),
)
def get_user_warnings(self) -> List[UserWarnings]:

2
kdb-bot/src/modules/base/base_module.py
View File

@ -38,6 +38,7 @@ from modules.base.events.base_on_voice_state_update_event_scheduled_event_bonus
from modules.base.helper.base_reaction_handler import BaseReactionHandler
from modules.base.service.base_helper_service import BaseHelperService
from modules.base.service.event_service import EventService
from modules.base.service.user_warnings_service import UserWarningsService
class BaseModule(ModuleABC):
@ -51,6 +52,7 @@ class BaseModule(ModuleABC):
services.add_transient(BaseHelperABC, BaseHelperService)
services.add_transient(BaseReactionHandler)
services.add_singleton(EventService)
services.add_transient(UserWarningsService)
# commands
self._dc.add_command(AFKCommand)

80
kdb-bot/src/modules/base/command/user_group.py
View File

@ -21,6 +21,8 @@ from bot_data.abc.user_joined_voice_channel_repository_abc import (
UserJoinedVoiceChannelRepositoryABC,
)
from bot_data.abc.user_repository_abc import UserRepositoryABC
from bot_data.abc.user_warnings_repository_abc import UserWarningsRepositoryABC
from modules.base.service.user_warnings_service import UserWarningsService
from modules.level.service.level_service import LevelService
from modules.permission.abc.permission_service_abc import PermissionServiceABC
@ -42,6 +44,8 @@ class UserGroup(DiscordCommandABC):
translate: TranslatePipe,
date: DateTimeOffsetPipe,
level: LevelService,
user_warnings: UserWarningsRepositoryABC,
user_warnings_service: UserWarningsService,
):
DiscordCommandABC.__init__(self)
@ -59,6 +63,8 @@ class UserGroup(DiscordCommandABC):
self._t = translate
self._date = date
self._level = level
self._user_warnings = user_warnings
self._user_warnings_service = user_warnings_service
self._logger.trace(__name__, f"Loaded command service: {type(self).__name__}")
@ -181,9 +187,13 @@ class UserGroup(DiscordCommandABC):
)
if is_mod or member == ctx.author:
warnings_string = ""
for warning in self._user_warnings.get_user_warnings_by_user_id(user.id):
warnings_string += f"{warning.id} - {warning.description}\n"
embed.add_field(
name=self._t.transform("modules.base.user.atr.warnings"),
value=self._t.transform("common.not_implemented_yet"),
value=warnings_string,
inline=False,
)
@ -341,3 +351,71 @@ class UserGroup(DiscordCommandABC):
self, interaction: discord.Interaction, current: str
) -> List[app_commands.Choice[str]]:
return [app_commands.Choice(name=value, value=key) for key, value in self._atr_list]
@user.group()
@commands.guild_only()
async def warning(self, ctx: Context):
pass
@warning.command()
@commands.guild_only()
@CommandChecks.check_is_ready()
@CommandChecks.check_is_member_moderator()
async def show(self, ctx: Context, member: discord.Member, wait: int = None):
self._logger.debug(__name__, f"Received command user warning show {ctx}:{member}")
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.id)
embed = discord.Embed(
title=member.name, description=self._t.transform("modules.base.user.atr.warnings"), color=int("ef9d0d", 16)
)
warnings_id_string = ""
for warning in self._user_warnings.get_user_warnings_by_user_id(user.id):
warnings_id_string += f"{warning.id}\n"
warnings_description_string = ""
for warning in self._user_warnings.get_user_warnings_by_user_id(user.id):
warnings_description_string += f"{warning.description}\n"
embed.add_field(
name=self._t.transform("modules.base.warnings.show.id"),
value=warnings_id_string,
inline=True,
)
embed.add_field(
name=self._t.transform("modules.base.warnings.show.description"),
value=warnings_description_string,
inline=True,
)
await self._message_service.send_interaction_msg(ctx.interaction, embed, wait_before_delete=wait)
self._logger.trace(__name__, f"Finished user warning show command")
@warning.command()
@commands.guild_only()
@CommandChecks.check_is_ready()
@CommandChecks.check_is_member_moderator()
async def add(self, ctx: Context, member: discord.Member, description: str):
self._logger.debug(__name__, f"Received command user warning add {ctx}:{member},{description}")
try:
await self._message_service.send_ctx_msg(ctx, self._t.transform("modules.base.warnings.add.success"))
await self._user_warnings_service.add_warnings(member, description, ctx.author.id)
except Exception as e:
self._logger.error(__name__, f"Adding user warning failed", e)
await self._message_service.send_ctx_msg(ctx, self._t.transform("modules.base.warnings.add.failed"))
self._logger.trace(__name__, f"Finished user warning add command")
@warning.command()
@commands.guild_only()
@CommandChecks.check_is_ready()
@CommandChecks.check_is_member_moderator()
async def remove(self, ctx: Context, warning_id: int):
self._logger.debug(__name__, f"Received command user warning remove {ctx}:{warning_id}")
try:
await self._message_service.send_ctx_msg(ctx, self._t.transform("modules.base.warnings.remove.success"))
await self._user_warnings_service.remove_warnings(warning_id)
except Exception as e:
self._logger.error(__name__, f"Removing user warning failed", e)
await self._message_service.send_ctx_msg(ctx, self._t.transform("modules.base.warnings.remove.failed"))
self._logger.trace(__name__, f"Finished user warning remove command")

6
kdb-bot/src/modules/base/configuration/base_server_settings.py
View File

@ -20,6 +20,7 @@ class BaseServerSettings(ConfigurationModelABC):
self._afk_command_channel_id: int = 0
self._help_command_reference_url: str = ""
self._help_voice_channel_id: int = 0
self._team_channel_id: int = 0
self._ping_urls = List(str)
@property
@ -62,6 +63,10 @@ class BaseServerSettings(ConfigurationModelABC):
def help_command_reference_url(self) -> str:
return self._help_command_reference_url
@property
def team_channel_id(self) -> int:
return self._team_channel_id
@property
def help_voice_channel_id(self) -> int:
return self._help_voice_channel_id
@ -86,6 +91,7 @@ class BaseServerSettings(ConfigurationModelABC):
self._afk_command_channel_id = settings["AFKCommandChannelId"]
self._help_command_reference_url = settings["HelpCommandReferenceUrl"]
self._help_voice_channel_id = settings["HelpVoiceChannelId"]
self._team_channel_id = settings["TeamChannelId"]
for url in settings["PingURLs"]:
self._ping_urls.append(url)
except Exception as e:

135
kdb-bot/src/modules/base/service/user_warnings_service.py
View File

@ -1,3 +1,4 @@
import discord
from cpl_core.database.context import DatabaseContextABC
from cpl_core.logging import LoggerABC
from cpl_discord.service import DiscordBotServiceABC
@ -7,8 +8,11 @@ from bot_core.abc.message_service_abc import MessageServiceABC
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.abc.user_warning_repository_abc import UserWarningsRepositoryABC
from bot_data.abc.user_warnings_repository_abc import UserWarningsRepositoryABC
from bot_data.model.user import User
from bot_data.model.user_warnings import UserWarnings
from modules.base.abc.base_helper_abc import BaseHelperABC
from modules.base.configuration.base_server_settings import BaseServerSettings
from modules.level.service.level_service import LevelService
from modules.permission.abc.permission_service_abc import PermissionServiceABC
@ -27,6 +31,7 @@ class UserWarningsService:
message_service: MessageServiceABC,
t: TranslatePipe,
permissions: PermissionServiceABC,
base_helper: BaseHelperABC,
):
self._logger = logger
self._db = db
@ -39,13 +44,70 @@ class UserWarningsService:
self._message_service = message_service
self._t = t
self._permissions = permissions
self._base_helper = base_helper
async def add_warnings(self, guild_id: int, member_id: int, description: str, author_id: int = None):
server = self._servers.get_server_by_discord_id(guild_id)
user = self._users.get_user_by_discord_id_and_server_id(member_id, server.id)
async def notify_team(self, member: discord.Member, description: str, removed=False):
try:
settings: BaseServerSettings = self._base_helper.get_config(member.guild.id)
channel = member.guild.get_channel(settings.team_channel_id)
if removed:
translation = self._t.transform("modules.base.warnings.team_removed").format(
description, member.mention
)
else:
translation = self._t.transform("modules.base.warnings.team_warned").format(member.mention, description)
guild = self._bot.get_guild(guild_id)
member = guild.get_member(member_id)
await self._message_service.send_channel_message(channel, translation)
except Exception as e:
self._logger.error(__name__, f"Team notification for user warning failed!", e)
async def notify_user(self, member: discord.Member, message: str):
try:
await self._message_service.send_dm_message(message, member)
except Exception as e:
self._logger.error(__name__, f"User notification for user warning failed!", e)
async def check_for_warnings(self, member: discord.Member, user: User):
existing_warnings = self._warnings.get_user_warnings_by_user_id(user.id)
if existing_warnings.count() == 1:
await self._message_service.send_dm_message(self._t.transform("modules.base.warnings.first"), member)
elif existing_warnings.count() == 2:
server = self._servers.get_server_by_discord_id(member.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)
new_level = levels.where(lambda l: l.min_xp < level.min_xp).last_or_default()
if new_level is not None:
user.xp = new_level.min_xp
self._users.update_user(user)
self._db.save_changes()
await self._message_service.send_dm_message(self._t.transform("modules.base.warnings.second"), member)
elif existing_warnings.count() == 3:
server = self._servers.get_server_by_discord_id(member.guild.id)
user = self._users.get_user_by_discord_id_and_server_id(member.id, server.id)
levels = self._levels.get_levels_by_server_id(server.id)
new_level = levels.where(lambda l: l.min_xp > 0).order_by(lambda l: l.min_xp).last_or_default()
if new_level is not None:
user.xp = new_level.min_xp
self._users.update_user(user)
self._db.save_changes()
await self._message_service.send_dm_message(self._t.transform("modules.base.warnings.third"), member)
elif existing_warnings.count() >= 4:
user.xp = 0
self._users.update_user(user)
self._db.save_changes()
await self.notify_team(member, self._t.transform("modules.base.warnings.kick").format(member.mention))
await member.kick()
async def add_warnings(self, member: discord.Member, description: str, author_id: int = None):
server = self._servers.get_server_by_discord_id(member.guild.id)
user = self._users.get_user_by_discord_id_and_server_id(member.id, server.id)
author = None
if author_id is not None:
@ -54,56 +116,17 @@ class UserWarningsService:
warning = UserWarnings(description, user, author)
self._warnings.add_user_warnings(warning)
self._db.save_changes()
await self.notify_user(member, self._t.transform("modules.base.warnings.warned").format(warning.description))
await self.notify_team(member, warning.description)
await self.check_for_warnings(member, user)
existing_warnings = self._warnings.get_user_warnings_by_user_id(user.id)
await self._message_service.send_dm_message(
self._t.transform("modules.base.warnings.message").format(warning.description), member
)
if existing_warnings.count() == 1:
await self._message_service.send_dm_message(
self._t.transform("modules.base.warnings.messages.first"), member
)
elif existing_warnings.count() == 2:
server = self._servers.get_server_by_discord_id(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)
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_dm_message(
self._t.transform("modules.base.warnings.messages.second"), member
)
elif existing_warnings.count() == 3:
server = self._servers.get_server_by_discord_id(guild_id)
user = self._users.get_user_by_discord_id_and_server_id(member_id, server.id)
levels = self._levels.get_levels_by_server_id(server.id)
new_level = levels.where(lambda l: l.min_xp > 0).order_by(lambda l: l.min_xp).last()
user.xp = new_level.min_xp
self._users.update_user(user)
self._db.save_changes()
await self._message_service.send_dm_message(
self._t.transform("modules.base.warnings.messages.third"), member
)
elif existing_warnings.count() >= 4:
user.xp = 0
self._users.update_user(user)
self._db.save_changes()
await member.kick()
mods = [
*self._permissions.get_admins(member.guild.id),
*self._permissions.get_moderators(member.guild.id),
]
for a in mods:
await self._message_service.send_dm_message(
self._t.transform("modules.base.warnings.messages.kick").format(member.mention),
a,
)
def remove_warnings(self, id: int):
async def remove_warnings(self, id: int):
warning = self._warnings.get_user_warnings_by_id(id)
self._warnings.delete_user_warnings(warning)
self._db.save_changes()
guild = self._bot.get_guild(warning.user.server.discord_id)
member = guild.get_member(warning.user.discord_id)
await self.notify_user(member, self._t.transform("modules.base.warnings.removed").format(warning.description))
await self.notify_team(member, warning.description, removed=True)