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

Removed statistic module #190

Browse Source
This commit is contained in:
Nick Jungmann 2023-02-17 00:02:26 +01:00
parent 64985f5983
commit 85e664e642
19 changed files with 25 additions and 961 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
kdb-bot/src/bot/bot.json
View File

@ -19,7 +19,7 @@
"cpl-core==2022.12.1.post3",
"cpl-translation==2022.12.1",
"cpl-query==2022.12.2.post1",
"Flask==2.2.2",
"Flask==2.2.3",
"Flask-Classful==0.14.2",
"Flask-Cors==3.0.10",
"PyJWT==2.6.0",
@ -41,15 +41,15 @@
"BuildSettings": {
"ProjectType": "console",
"SourcePath": "",
"OutputPath": "../../dist",
"OutputPath": "..\\..\\dist",
"Main": "bot.main",
"EntryPoint": "bot",
"IncludePackageData": false,
"Included": [],
"Excluded": [
"*/__pycache__",
"*/logs",
"*/tests"
"*\\__pycache__",
"*\\logs",
"*\\tests"
],
"PackageData": {},
"ProjectReferences": [

2
kdb-bot/src/bot/module_list.py
View File

@ -11,7 +11,6 @@ from modules.boot_log.boot_log_module import BootLogModule
from modules.database.database_module import DatabaseModule
from modules.level.level_module import LevelModule
from modules.permission.permission_module import PermissionModule
from modules.stats.stats_module import StatsModule
from modules.technician.technician_module import TechnicianModule
@ -31,7 +30,6 @@ class ModuleList:
BaseModule,
LevelModule,
ApiModule,
StatsModule,
TechnicianModule,
# has to be last!
BootLogModule,

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

@ -263,30 +263,6 @@
},
"database": {},
"permission": {},
"stats": {
"list": {
"statistic": "Statistik",
"description": "Beschreibung",
"nothing_found": "Keine Statistiken gefunden."
},
"view": {
"statistic": "Statistik",
"description": "Beschreibung",
"failed": "Statistik kann nicht gezeigt werden :("
},
"add": {
"failed": "Statistik kann nicht hinzugefügt werden :(",
"success": "Statistik wurde hinzugefügt :D"
},
"edit": {
"failed": "Statistik kann nicht bearbeitet werden :(",
"success": "Statistik wurde gespeichert :D"
},
"remove": {
"failed": "Statistik kann nicht gelöscht werden :(",
"success": "Statistik wurde gelöscht :D"
}
},
"technician": {
"restart_message": "Bin gleich wieder da :D",
"shutdown_message": "Trauert nicht um mich, es war eine logische Entscheidung. Das Wohl von Vielen, es wiegt schwerer als das Wohl von Wenigen oder eines Einzelnen. Ich war es und ich werde es immer sein, Euer Freund. Lebt lange und in Frieden :)",

44
kdb-bot/src/bot_data/abc/statistic_repository_abc.py
View File

@ -1,44 +0,0 @@
from abc import ABC, abstractmethod
from typing import Optional
from cpl_query.extension import List
from bot_data.model.statistic import Statistic
class StatisticRepositoryABC(ABC):
@abstractmethod
def __init__(self):
pass
@abstractmethod
def get_statistics(self) -> List[Statistic]:
pass
@abstractmethod
def get_statistics_by_server_id(self, server_id: int) -> List[Statistic]:
pass
@abstractmethod
def get_statistic_by_id(self, id: int) -> Statistic:
pass
@abstractmethod
def get_statistic_by_name(self, name: str, server_id: int) -> Statistic:
pass
@abstractmethod
def find_statistic_by_name(self, name: str, server_id: int) -> Optional[Statistic]:
pass
@abstractmethod
def add_statistic(self, statistic: Statistic):
pass
@abstractmethod
def update_statistic(self, statistic: Statistic):
pass
@abstractmethod
def delete_statistic(self, statistic: Statistic):
pass

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

@ -12,7 +12,6 @@ from bot_data.abc.client_repository_abc import ClientRepositoryABC
from bot_data.abc.known_user_repository_abc import KnownUserRepositoryABC
from bot_data.abc.level_repository_abc import LevelRepositoryABC
from bot_data.abc.server_repository_abc import ServerRepositoryABC
from bot_data.abc.statistic_repository_abc import StatisticRepositoryABC
from bot_data.abc.user_joined_game_server_repository_abc import UserJoinedGameServerRepositoryABC
from bot_data.abc.user_joined_server_repository_abc import UserJoinedServerRepositoryABC
from bot_data.abc.user_joined_voice_channel_repository_abc import (
@ -30,7 +29,6 @@ from bot_data.service.known_user_repository_service import KnownUserRepositorySe
from bot_data.service.level_repository_service import LevelRepositoryService
from bot_data.service.seeder_service import SeederService
from bot_data.service.server_repository_service import ServerRepositoryService
from bot_data.service.statistic_repository_service import StatisticRepositoryService
from bot_data.service.user_joined_game_server_repository_service import UserJoinedGameServerRepositoryService
from bot_data.service.user_joined_server_repository_service import (
UserJoinedServerRepositoryService,
@ -63,7 +61,6 @@ class DataModule(ModuleABC):
services.add_transient(UserJoinedGameServerRepositoryABC, UserJoinedGameServerRepositoryService)
services.add_transient(AutoRoleRepositoryABC, AutoRoleRepositoryService)
services.add_transient(LevelRepositoryABC, LevelRepositoryService)
services.add_transient(StatisticRepositoryABC, StatisticRepositoryService)
services.add_transient(
UserMessageCountPerHourRepositoryABC,
UserMessageCountPerHourRepositoryService,

34
kdb-bot/src/bot_data/migration/stats_migration.py
View File

@ -4,7 +4,7 @@ from bot_data.db_context import DBContext
class StatsMigration(MigrationABC):
name = "0.3_StatsMigration"
name = "1.0_RemoveStatsMigration"
def __init__(self, logger: DatabaseLogger, db: DBContext):
MigrationABC.__init__(self)
@ -18,20 +18,26 @@ class StatsMigration(MigrationABC):
self._cursor.execute(
str(
f"""
CREATE TABLE IF NOT EXISTS `Statistics` (
`Id` BIGINT NOT NULL AUTO_INCREMENT,
`Name` VARCHAR(255) NOT NULL,
`Description` VARCHAR(255) NOT NULL,
`Code` LONGTEXT NOT NULL,
`ServerId` BIGINT,
`CreatedAt` DATETIME(6),
`LastModifiedAt` DATETIME(6),
PRIMARY KEY(`Id`),
FOREIGN KEY (`ServerId`) REFERENCES `Servers`(`ServerId`)
);
"""
DROP TABLE IF EXISTS `Statistics`;
"""
)
)
def downgrade(self):
self._cursor.execute("DROP TABLE `Statistics`;")
self._cursor.execute(
str(
f"""
CREATE TABLE IF NOT EXISTS `Statistics` (
`Id` BIGINT NOT NULL AUTO_INCREMENT,
`Name` VARCHAR(255) NOT NULL,
`Description` VARCHAR(255) NOT NULL,
`Code` LONGTEXT NOT NULL,
`ServerId` BIGINT,
`CreatedAt` DATETIME(6),
`LastModifiedAt` DATETIME(6),
PRIMARY KEY(`Id`),
FOREIGN KEY (`ServerId`) REFERENCES `Servers`(`ServerId`)
);
"""
)
)

131
kdb-bot/src/bot_data/model/statistic.py
View File

@ -1,131 +0,0 @@
from datetime import datetime
from cpl_core.database import TableABC
from cpl_core.utils import CredentialManager
from bot_data.model.server import Server
class Statistic(TableABC):
def __init__(
self,
name: str,
description: str,
code: str,
server: Server,
created_at: datetime = None,
modified_at: datetime = None,
id=0,
):
self._id = id
self._name = name
self._description = description
self._code = CredentialManager.encrypt(code)
self._server = server
TableABC.__init__(self)
self._created_at = created_at if created_at is not None else self._created_at
self._modified_at = modified_at if modified_at is not None else self._modified_at
@property
def id(self) -> int:
return self._id
@property
def name(self) -> str:
return self._name
@property
def description(self) -> str:
return self._description
@description.setter
def description(self, value: str):
self._description = value
@property
def code(self) -> str:
return CredentialManager.decrypt(self._code)
@code.setter
def code(self, value: str):
self._code = CredentialManager.encrypt(value)
@property
def server(self) -> Server:
return self._server
@staticmethod
def get_select_all_string() -> str:
return str(
f"""
SELECT * FROM `Statistics`;
"""
)
@staticmethod
def get_select_by_id_string(id: int) -> str:
return str(
f"""
SELECT * FROM `Statistics`
WHERE `Id` = {id};
"""
)
@staticmethod
def get_select_by_name_string(name: str, s_id: int) -> str:
return str(
f"""
SELECT * FROM `Statistics`
WHERE `ServerId` = {s_id}
AND `Name` = '{name}';
"""
)
@staticmethod
def get_select_by_server_string(s_id: int) -> str:
return str(
f"""
SELECT * FROM `Statistics`
WHERE `ServerId` = {s_id};
"""
)
@property
def insert_string(self) -> str:
return str(
f"""
INSERT INTO `Statistics` (
`Name`, `Description`, `Code`, `ServerId`, `CreatedAt`, `LastModifiedAt`
) VALUES (
'{self._name}',
'{self._description}',
'{self._code}',
{self._server.id},
'{self._created_at}',
'{self._modified_at}'
);
"""
)
@property
def udpate_string(self) -> str:
return str(
f"""
UPDATE `Statistics`
SET `Name` = '{self._name}',
`Description` = '{self._description}',
`Code` = '{self._code}',
`LastModifiedAt` = '{self._modified_at}'
WHERE `Id` = {self._id};
"""
)
@property
def delete_string(self) -> str:
return str(
f"""
DELETE FROM `Statistics`
WHERE `Id` = {self._id};
"""
)

106
kdb-bot/src/bot_data/service/statistic_repository_service.py
View File

@ -1,106 +0,0 @@
from typing import Optional
from cpl_core.database.context import DatabaseContextABC
from cpl_core.utils import CredentialManager
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.statistic_repository_abc import StatisticRepositoryABC
from bot_data.model.statistic import Statistic
class StatisticRepositoryService(StatisticRepositoryABC):
def __init__(
self,
logger: DatabaseLogger,
db_context: DatabaseContextABC,
statistics: ServerRepositoryABC,
):
self._logger = logger
self._context = db_context
self._statistics = statistics
StatisticRepositoryABC.__init__(self)
@staticmethod
def _get_value_from_result(value: any) -> Optional[any]:
if isinstance(value, str) and "NULL" in value:
return None
return value
def _statistic_from_result(self, sql_result: tuple) -> Statistic:
code = self._get_value_from_result(sql_result[3])
if code is not None:
code = CredentialManager.decrypt(code)
statistic = Statistic(
self._get_value_from_result(sql_result[1]),
self._get_value_from_result(sql_result[2]),
code,
self._statistics.get_server_by_id(sql_result[4]),
id=self._get_value_from_result(sql_result[0]),
)
return statistic
def get_statistics(self) -> List[Statistic]:
statistics = List(Statistic)
self._logger.trace(__name__, f"Send SQL command: {Statistic.get_select_all_string()}")
results = self._context.select(Statistic.get_select_all_string())
for result in results:
statistics.append(self._statistic_from_result(result))
return statistics
def get_statistics_by_server_id(self, server_id: int) -> List[Statistic]:
statistics = List(Statistic)
self._logger.trace(
__name__,
f"Send SQL command: {Statistic.get_select_by_server_string(server_id)}",
)
results = self._context.select(Statistic.get_select_by_server_string(server_id))
for result in results:
statistics.append(self._statistic_from_result(result))
return statistics
def get_statistic_by_id(self, id: int) -> Statistic:
self._logger.trace(__name__, f"Send SQL command: {Statistic.get_select_by_id_string(id)}")
result = self._context.select(Statistic.get_select_by_id_string(id))[0]
return self._statistic_from_result(result)
def get_statistic_by_name(self, name: str, server_id: int) -> Statistic:
self._logger.trace(
__name__,
f"Send SQL command: {Statistic.get_select_by_name_string(name, server_id)}",
)
result = self._context.select(Statistic.get_select_by_name_string(name, server_id))[0]
return self._statistic_from_result(result)
def find_statistic_by_name(self, name: str, server_id: int) -> Optional[Statistic]:
self._logger.trace(
__name__,
f"Send SQL command: {Statistic.get_select_by_name_string(name, server_id)}",
)
result = self._context.select(Statistic.get_select_by_name_string(name, server_id))
if result is None or len(result) == 0:
return None
result = result[0]
return self._statistic_from_result(result)
def add_statistic(self, statistic: Statistic):
self._logger.trace(__name__, f"Send SQL command: {statistic.insert_string}")
self._context.cursor.execute(statistic.insert_string)
def update_statistic(self, statistic: Statistic):
self._logger.trace(__name__, f"Send SQL command: {statistic.udpate_string}")
self._context.cursor.execute(statistic.udpate_string)
def delete_statistic(self, statistic: Statistic):
self._logger.trace(__name__, f"Send SQL command: {statistic.delete_string}")
self._context.cursor.execute(statistic.delete_string)

26
kdb-bot/src/modules/stats/__init__.py
View File

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

26
kdb-bot/src/modules/stats/command/__init__.py
View File

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

232
kdb-bot/src/modules/stats/command/stats_group.py
View File

@ -1,232 +0,0 @@
from typing import List as TList
import discord
from cpl_core.database.context import DatabaseContextABC
from cpl_discord.command import DiscordCommandABC
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.helper.command_checks import CommandChecks
from bot_core.logging.command_logger import CommandLogger
from bot_data.abc.server_repository_abc import ServerRepositoryABC
from bot_data.abc.statistic_repository_abc import StatisticRepositoryABC
from modules.permission.abc.permission_service_abc import PermissionServiceABC
from modules.stats.service.statistic_service import StatisticService
from modules.stats.ui.add_statistic_form import AddStatisticForm
class StatsGroup(DiscordCommandABC):
def __init__(
self,
logger: CommandLogger,
message_service: MessageServiceABC,
client_utils: ClientUtilsABC,
translate: TranslatePipe,
permission_service: PermissionServiceABC,
statistic: StatisticService,
servers: ServerRepositoryABC,
stats: StatisticRepositoryABC,
db: DatabaseContextABC,
):
DiscordCommandABC.__init__(self)
self._logger = logger
self._client_utils = client_utils
self._message_service = message_service
self._t = translate
self._permissions = permission_service
self._statistic = statistic
self._servers = servers
self._stats = stats
self._db = db
@commands.hybrid_group()
@commands.guild_only()
async def stats(self, ctx: Context):
pass
@stats.command()
@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 stats list {ctx}")
if ctx.guild is None:
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),
)
server = self._servers.get_server_by_discord_id(ctx.guild.id)
stats = self._stats.get_statistics_by_server_id(server.id)
if stats.count() == 0:
await self._message_service.send_ctx_msg(ctx, self._t.transform("modules.stats.list.nothing_found"))
return
statistics = ""
descriptions = ""
for statistic in stats:
statistics += f"\n{statistic.name}"
descriptions += f"\n{statistic.description}"
embed.add_field(
name=self._t.transform("modules.stats.list.statistic"),
value=statistics,
inline=True,
)
embed.add_field(
name=self._t.transform("modules.stats.list.description"),
value=descriptions,
inline=True,
)
await self._message_service.send_ctx_msg(ctx, embed, wait_before_delete=wait)
self._logger.trace(__name__, f"Finished command stats list")
@stats.command()
@commands.guild_only()
@CommandChecks.check_is_ready()
@CommandChecks.check_is_member_moderator()
async def view(self, ctx: Context, name: str, wait: int = None):
self._logger.debug(__name__, f"Received command stats view {ctx}:{name}")
if ctx.guild is None:
return
try:
server = self._servers.get_server_by_discord_id(ctx.guild.id)
stats = self._stats.get_statistics_by_server_id(server.id)
statistic = stats.where(lambda s: s.name == name).single()
result = await self._statistic.execute(statistic.code, server)
embed = discord.Embed(
title=statistic.name,
description=statistic.description,
color=int("ef9d0d", 16),
)
for i in range(result.header.count()):
header = result.header[i]
value = ""
for row in result.values:
value += f"\n{row[i]}"
embed.add_field(name=header, value=value, inline=True)
await self._message_service.send_ctx_msg(ctx, embed, wait_before_delete=wait)
except Exception as e:
self._logger.error(__name__, f"Cannot view statistic {name}", e)
await self._message_service.send_ctx_msg(ctx, self._t.transform("modules.stats.view.failed"))
self._logger.trace(__name__, f"Finished stats view command")
@view.autocomplete("name")
async def view_autocomplete(
self, interaction: discord.Interaction, current: str
) -> TList[app_commands.Choice[str]]:
server = self._servers.get_server_by_discord_id(interaction.guild.id)
stats = self._stats.get_statistics_by_server_id(server.id)
return [
app_commands.Choice(name=f"{statistic.name}: {statistic.description}", value=statistic.name)
for statistic in stats
]
@stats.command()
@commands.guild_only()
@CommandChecks.check_is_ready()
@CommandChecks.check_is_member_technician()
async def add(self, ctx: Context, name: str):
self._logger.debug(__name__, f"Received command stats add {ctx}: {name}")
if ctx.guild is None:
return
server = self._servers.get_server_by_discord_id(ctx.guild.id)
form = AddStatisticForm(
server,
self._stats,
self._db,
name,
self._message_service,
self._logger,
self._t,
)
self._logger.trace(__name__, f"Finished stats add command")
self._logger.trace(__name__, f"Started stats command form")
await ctx.interaction.response.send_modal(form)
@stats.command()
@commands.guild_only()
@CommandChecks.check_is_ready()
@CommandChecks.check_is_member_technician()
async def edit(self, ctx: Context, name: str):
self._logger.debug(__name__, f"Received command stats edit {ctx}: {name}")
try:
server = self._servers.get_server_by_discord_id(ctx.guild.id)
stats = self._stats.get_statistics_by_server_id(server.id)
statistic = stats.where(lambda s: s.name == name).single()
form = AddStatisticForm(
server,
self._stats,
self._db,
name,
self._message_service,
self._logger,
self._t,
code=statistic.code,
description=statistic.description,
)
self._logger.trace(__name__, f"Finished stats edit command")
self._logger.trace(__name__, f"Started stats command form")
await ctx.interaction.response.send_modal(form)
except Exception as e:
self._logger.error(__name__, f"Cannot edit statistic {name}", e)
await self._message_service.send_ctx_msg(ctx, self._t.transform("modules.stats.edit.failed"))
@edit.autocomplete("name")
async def edit_autocomplete(
self, interaction: discord.Interaction, current: str
) -> TList[app_commands.Choice[str]]:
server = self._servers.get_server_by_discord_id(interaction.guild.id)
stats = self._stats.get_statistics_by_server_id(server.id)
return [
app_commands.Choice(name=f"{statistic.name}: {statistic.description}", value=statistic.name)
for statistic in stats
]
@stats.command()
@commands.guild_only()
@CommandChecks.check_is_ready()
@CommandChecks.check_is_member_technician()
async def remove(self, ctx: Context, name: str):
self._logger.debug(__name__, f"Received command stats remove {ctx}: {name}")
try:
server = self._servers.get_server_by_discord_id(ctx.guild.id)
statistic = self._stats.get_statistic_by_name(name, server.id)
self._stats.delete_statistic(statistic)
self._db.save_changes()
await self._message_service.send_ctx_msg(ctx, self._t.transform("modules.stats.remove.success"))
self._logger.trace(__name__, f"Finished stats remove command")
except Exception as e:
self._logger.error(__name__, f"Cannot remove statistic {name}", e)
await self._message_service.send_ctx_msg(ctx, self._t.transform("modules.stats.remove.failed"))
@remove.autocomplete("name")
async def edit_autocomplete(
self, interaction: discord.Interaction, current: str
) -> TList[app_commands.Choice[str]]:
server = self._servers.get_server_by_discord_id(interaction.guild.id)
stats = self._stats.get_statistics_by_server_id(server.id)
return [
app_commands.Choice(name=f"{statistic.name}: {statistic.description}", value=statistic.name)
for statistic in stats
]

26
kdb-bot/src/modules/stats/model/__init__.py
View File

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

23
kdb-bot/src/modules/stats/model/statistic_result.py
View File

@ -1,23 +0,0 @@
from cpl_query.extension import List
class StatisticResult:
def __init__(self):
self._header = List(str)
self._values = List(List)
@property
def header(self) -> List[str]:
return self._header
@header.setter
def header(self, value: List[str]):
self._header = value
@property
def values(self) -> List[List]:
return self._values
@values.setter
def values(self, value: List[List]):
self._values = value

26
kdb-bot/src/modules/stats/service/__init__.py
View File

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

83
kdb-bot/src/modules/stats/service/statistic_service.py
View File

@ -1,83 +0,0 @@
from cpl_discord.service import DiscordBotServiceABC
from cpl_query.extension import List
from discord import Guild
from bot_data.abc.auto_role_repository_abc import AutoRoleRepositoryABC
from bot_data.abc.client_repository_abc import ClientRepositoryABC
from bot_data.abc.known_user_repository_abc import KnownUserRepositoryABC
from bot_data.abc.level_repository_abc import LevelRepositoryABC
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_repository_abc import (
UserJoinedVoiceChannelRepositoryABC,
)
from bot_data.abc.user_repository_abc import UserRepositoryABC
from bot_data.model.auto_role import AutoRole
from bot_data.model.client import Client
from bot_data.model.known_user import KnownUser
from bot_data.model.level import Level
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 modules.stats.model.statistic_result import StatisticResult
class StatisticService:
def __init__(
self,
auto_roles: AutoRoleRepositoryABC,
clients: ClientRepositoryABC,
known_users: KnownUserRepositoryABC,
levels: LevelRepositoryABC,
servers: ServerRepositoryABC,
user_joined_servers: UserJoinedServerRepositoryABC,
user_joined_voice_channel: UserJoinedVoiceChannelRepositoryABC,
users: UserRepositoryABC,
bot: DiscordBotServiceABC,
):
self._auto_roles = auto_roles
self._clients = clients
self._known_users = known_users
self._levels = levels
self._servers = servers
self._user_joined_servers = user_joined_servers
self._user_joined_voice_channel = user_joined_voice_channel
self._users = users
self._bot = bot
async def execute(self, code: str, server: Server) -> StatisticResult:
guild = self._bot.guilds.where(lambda g: g.id == server.discord_id).single()
return await self.get_data(
code,
self._auto_roles.get_auto_roles().where(lambda x: x.server.id == server.id),
self._clients.get_clients().where(lambda x: x.server.id == server.id),
self._known_users.get_users(),
self._levels.get_levels().where(lambda x: x.server.id == server.id),
self._servers.get_servers().where(lambda x: x.id == server.id),
self._user_joined_servers.get_user_joined_servers().where(lambda x: x.user.server.id == server.id),
self._user_joined_voice_channel.get_user_joined_voice_channels().where(
lambda x: x.user.server.id == server.id
),
self._users.get_users().where(lambda x: x.server.id == server.id),
guild,
)
@staticmethod
async def get_data(
code: str,
auto_roles: List[AutoRole],
clients: List[Client],
known_users: List[KnownUser],
levels: List[Level],
servers: List[Server],
user_joined_servers: List[UserJoinedServer],
user_joined_voice_channel: List[UserJoinedVoiceChannel],
users: List[User],
guild: Guild,
) -> StatisticResult:
result = StatisticResult()
exec(code)
return result

44
kdb-bot/src/modules/stats/stats.json
View File

@ -1,44 +0,0 @@
{
"ProjectSettings": {
"Name": "stats",
"Version": {
"Major": "1",
"Minor": "0",
"Micro": "dev130"
},
"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": "stats.main",
"EntryPoint": "stats",
"IncludePackageData": false,
"Included": [],
"Excluded": [
"*/__pycache__",
"*/logs",
"*/tests"
],
"PackageData": {},
"ProjectReferences": []
}
}

23
kdb-bot/src/modules/stats/stats_module.py
View File

@ -1,23 +0,0 @@
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.stats.command.stats_group import StatsGroup
from modules.stats.service.statistic_service import StatisticService
class StatsModule(ModuleABC):
def __init__(self, dc: DiscordCollectionABC):
ModuleABC.__init__(self, dc, FeatureFlagsEnum.stats_module)
def configure_configuration(self, config: ConfigurationABC, env: ApplicationEnvironmentABC):
pass
def configure_services(self, services: ServiceCollectionABC, env: ApplicationEnvironmentABC):
services.add_transient(StatisticService)
# commands
self._dc.add_command(StatsGroup)
# events

26
kdb-bot/src/modules/stats/ui/__init__.py
View File

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

97
kdb-bot/src/modules/stats/ui/add_statistic_form.py
View File

@ -1,97 +0,0 @@
import discord
from cpl_core.database.context import DatabaseContextABC
from cpl_translation import TranslatePipe
from discord import ui, TextStyle
from bot_core.abc.message_service_abc import MessageServiceABC
from bot_core.logging.command_logger import CommandLogger
from bot_data.abc.statistic_repository_abc import StatisticRepositoryABC
from bot_data.model.server import Server
from bot_data.model.statistic import Statistic
class AddStatisticForm(ui.Modal):
description = ui.TextInput(label="Beschreibung", required=True)
code = ui.TextInput(label="Code", required=True, style=TextStyle.long)
def __init__(
self,
server: Server,
stats: StatisticRepositoryABC,
db: DatabaseContextABC,
name: str,
message_service: MessageServiceABC,
logger: CommandLogger,
t: TranslatePipe,
code: str = None,
description: str = None,
):
ui.Modal.__init__(self, title=name)
self._server = server
self._stats = stats
self._db = db
self._name = name
self._message_service = message_service
self._logger = logger
self._t = t
if code is not None:
self.code.default = code
if description is not None:
self.description.default = description
async def on_submit(self, interaction: discord.Interaction):
statistic = (
self._stats.get_statistics_by_server_id(self._server.id)
.where(lambda s: s.name == self._name)
.single_or_default()
)
if interaction.guild is None:
if statistic is None:
await self._message_service.send_interaction_msg(
interaction, self._t.transform("modules.stats.add.failed")
)
else:
await self._message_service.send_interaction_msg(
interaction, self._t.transform("modules.stats.edit.failed")
)
return
try:
if statistic is None:
self._stats.add_statistic(
Statistic(
self._name,
self.description.value,
self.code.value,
self._server,
)
)
self._db.save_changes()
await self._message_service.send_interaction_msg(
interaction, self._t.transform("modules.stats.add.success")
)
return
statistic.description = self.description.value
statistic.code = self.code.value
self._stats.update_statistic(statistic)
self._db.save_changes()
await self._message_service.send_interaction_msg(
interaction, self._t.transform("modules.stats.edit.success")
)
except Exception as e:
self._logger.error(__name__, f"Save statistic {self._name} failed", e)
if statistic is None:
await self._message_service.send_interaction_msg(
interaction, self._t.transform("modules.stats.add.failed")
)
else:
await self._message_service.send_interaction_msg(
interaction, self._t.transform("modules.stats.edit.failed")
)
self._logger.trace(__name__, f"Finished stats command form")