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

dev into staging #450

Merged
edraft merged 14 commits from dev into staging 2023-12-03 16:08:27 +01:00
Conversation 0 Commits 14 Files Changed 178 +1113 -1513
178 changed files with 1113 additions and 1513 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
bot/cpl-workspace.json
View File

@ -14,7 +14,6 @@
"config": "src/modules/config/config.json",
"database": "src/modules/database/database.json",
"level": "src/modules/level/level.json",
"permission": "src/modules/permission/permission.json",
"technician": "src/modules/technician/technician.json",
"short-role-name": "src/modules/short_role_name/short-role-name.json",
"special-offers": "src/modules/special_offers/special-offers.json",

4
bot/src/bot/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports:
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

3
bot/src/bot/bot.json
View File

@ -4,7 +4,7 @@
"Version": {
"Major": "1",
"Minor": "2",
"Micro": "2"
"Micro": "3"
},
"Author": "Sven Heidemann",
"AuthorEmail": "sven.heidemann@sh-edraft.de",
@ -70,7 +70,6 @@
"../modules/config/config.json",
"../modules/database/database.json",
"../modules/level/level.json",
"../modules/permission/permission.json",
"../modules/short_role_name/short-role-name.json",
"../modules/special_offers/special-offers.json",
"../modules/technician/technician.json"

4
bot/src/bot/extension/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot.extension"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports:
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

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

@ -12,7 +12,6 @@ from modules.boot_log.boot_log_module import BootLogModule
from modules.config.config_module import ConfigModule
from modules.database.database_module import DatabaseModule
from modules.level.level_module import LevelModule
from modules.permission.permission_module import PermissionModule
from modules.short_role_name.short_role_name_module import ShortRoleNameModule
from modules.special_offers.special_offers_module import SteamSpecialOffersModule
from modules.technician.technician_module import TechnicianModule
@ -30,7 +29,6 @@ class ModuleList:
ConfigModule, # has to be before db check
DatabaseModule,
GraphQLModule,
PermissionModule,
AutoRoleModule,
BaseModule,
LevelModule,

4
bot/src/bot_api/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_api"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports:
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

4
bot/src/bot_api/abc/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_api.abc"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports:
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

2
bot/src/bot_api/bot-api.json
View File

@ -4,7 +4,7 @@
"Version": {
"Major": "1",
"Minor": "2",
"Micro": "2"
"Micro": "3"
},
"Author": "",
"AuthorEmail": "",

4
bot/src/bot_api/configuration/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_api.configuration"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

4
bot/src/bot_api/controller/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_api.controller"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports:
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

4
bot/src/bot_api/event/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_api.event"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports:
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

4
bot/src/bot_api/exception/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_api.exception"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports:
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

4
bot/src/bot_api/filter/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_api.filter"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports:
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

4
bot/src/bot_api/filter/discord/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_api.filter.discord"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports:
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

4
bot/src/bot_api/logging/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_api.logging"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports:
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

4
bot/src/bot_api/model/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_api.model"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports:
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

4
bot/src/bot_api/model/discord/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_api.model.discord"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports:
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

4
bot/src/bot_api/route/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_api.route"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports:
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

4
bot/src/bot_api/service/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_api.service"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

4
bot/src/bot_api/transformer/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_api.transformer"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports:
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

2
bot/src/bot_api/transformer/auth_user_transformer.py
View File

@ -10,7 +10,7 @@ from bot_api.model.user_dto import UserDTO
from bot_data.model.auth_role_enum import AuthRoleEnum
from bot_data.model.auth_user import AuthUser
from bot_data.model.user import User
from modules.permission.abc.permission_service_abc import PermissionServiceABC
from bot_core.abc.permission_service_abc import PermissionServiceABC
class AuthUserTransformer(TransformerABC):

4
bot/src/bot_core/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_core"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

4
bot/src/bot_core/abc/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_core.abc"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports:
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

10
bot/src/bot_core/abc/client_utils_abc.py
View File

@ -48,6 +48,16 @@ class ClientUtilsABC(ABC):
def get_auto_complete_list(self, _l: List, current: str, select: Callable = None) -> List:
pass
@abstractmethod
def update_user_message_xp_count_by_hour(
self,
created_at: datetime,
user: User,
settings: ServerConfig,
is_reaction: bool = False,
):
pass
@abstractmethod
def is_message_xp_count_by_hour_higher_that_max_message_count_per_hour(
self,

0
bot/src/modules/permission/abc/permission_service_abc.py → bot/src/bot_core/abc/permission_service_abc.py
View File

2
bot/src/bot_core/bot-core.json
View File

@ -4,7 +4,7 @@
"Version": {
"Major": "1",
"Minor": "2",
"Micro": "2"
"Micro": "3"
},
"Author": "Sven Heidemann",
"AuthorEmail": "sven.heidemann@sh-edraft.de",

4
bot/src/bot_core/configuration/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_core.configuration"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports:
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

1
bot/src/bot_core/configuration/feature_flags_enum.py
View File

@ -15,7 +15,6 @@ class FeatureFlagsEnum(Enum):
database_module = "DatabaseModule"
level_module = "LevelModule"
moderator_module = "ModeratorModule"
permission_module = "PermissionModule"
short_role_name_module = "ShortRoleNameModule"
steam_special_offers_module = "SteamSpecialOffersModule"
# features

1
bot/src/bot_core/configuration/feature_flags_settings.py
View File

@ -16,7 +16,6 @@ class FeatureFlagsSettings(ConfigurationModelABC):
FeatureFlagsEnum.data_module.value: True, # 03.10.2022 #56
FeatureFlagsEnum.database_module.value: True, # 02.10.2022 #48
FeatureFlagsEnum.moderator_module.value: False, # 02.10.2022 #48
FeatureFlagsEnum.permission_module.value: True, # 02.10.2022 #48
FeatureFlagsEnum.config_module.value: True, # 19.07.2023 #127
FeatureFlagsEnum.short_role_name_module.value: True, # 28.09.2023 #378
FeatureFlagsEnum.steam_special_offers_module.value: True, # 11.10.2023 #188

4
bot/src/bot_core/core_extension/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_core.core_extension"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports:
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

2
bot/src/bot_core/core_extension/core_extension.py
View File

@ -9,7 +9,7 @@ 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.helper.event_checks import EventChecks
from modules.permission.abc.permission_service_abc import PermissionServiceABC
from bot_core.abc.permission_service_abc import PermissionServiceABC
class CoreExtension(ApplicationExtensionABC):

3
bot/src/bot_core/core_module.py
View File

@ -7,6 +7,7 @@ from cpl_discord.service.discord_collection_abc import DiscordCollectionABC
from bot_core.abc.client_utils_abc import ClientUtilsABC
from bot_core.abc.message_service_abc import MessageServiceABC
from bot_core.abc.module_abc import ModuleABC
from bot_core.abc.permission_service_abc import PermissionServiceABC
from bot_core.configuration.feature_flags_enum import FeatureFlagsEnum
from bot_core.events.core_on_ready_event import CoreOnReadyEvent
from bot_core.pipes.date_time_offset_pipe import DateTimeOffsetPipe
@ -14,6 +15,7 @@ from bot_core.service.client_utils_service import ClientUtilsService
from bot_core.service.config_service import ConfigService
from bot_core.service.data_integrity_service import DataIntegrityService
from bot_core.service.message_service import MessageService
from bot_core.service.permission_service import PermissionService
class CoreModule(ModuleABC):
@ -28,6 +30,7 @@ class CoreModule(ModuleABC):
services.add_transient(MessageServiceABC, MessageService)
services.add_transient(ClientUtilsABC, ClientUtilsService)
services.add_transient(DataIntegrityService)
services.add_singleton(PermissionServiceABC, PermissionService)
# pipes
services.add_transient(DateTimeOffsetPipe)

4
bot/src/bot_core/events/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_core.events"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports:
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

4
bot/src/bot_core/exception/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_core.exception"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports:
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

4
bot/src/bot_core/helper/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_core.helper"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports:
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

2
bot/src/bot_core/helper/command_checks.py
View File

@ -7,7 +7,7 @@ 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.exception.check_error import CheckError
from modules.permission.abc.permission_service_abc import PermissionServiceABC
from bot_core.abc.permission_service_abc import PermissionServiceABC
class CommandChecks:

4
bot/src/bot_core/logging/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_core.logging"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

4
bot/src/bot_core/pipes/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_core.pipes"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

4
bot/src/bot_core/service/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_core.service"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports:
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

57
bot/src/bot_core/service/client_utils_service.py
View File

@ -143,14 +143,13 @@ class ClientUtilsService(ClientUtilsABC):
return _l.take(25)
def is_message_xp_count_by_hour_higher_that_max_message_count_per_hour(
def update_user_message_xp_count_by_hour(
self,
created_at: datetime,
user: User,
settings: ServerConfig,
is_reaction: bool = False,
) -> bool:
umcph = None
):
try:
umcph = self._umcphs.find_user_message_count_per_hour_by_user_id_and_date(user.id, created_at)
if umcph is None:
@ -162,44 +161,50 @@ class ClientUtilsService(ClientUtilsABC):
user,
)
)
self._db.save_changes()
umcph = self._umcphs.get_user_message_count_per_hour_by_user_id_and_date(user.id, created_at)
except Exception as e:
self._logger.error(
__name__,
f"Cannot add user message count per hour with id {umcph.id}",
e,
)
return False
try:
if is_reaction:
umcph.xp_count += settings.xp_per_reaction
else:
umcph.xp_count += settings.xp_per_message
umcph.xp_count += settings.xp_per_reaction if is_reaction else settings.xp_per_message
self._umcphs.update_user_message_count_per_hour(umcph)
self._db.save_changes()
except Exception as e:
self._logger.error(
__name__,
f"Cannot update user message count per hour with id {umcph.id}",
f"Cannot update user message count per hour {created_at}",
e,
)
return False
if umcph.xp_count is None:
return False
def is_message_xp_count_by_hour_higher_that_max_message_count_per_hour(
self,
created_at: datetime,
user: User,
settings: ServerConfig,
is_reaction: bool = False,
) -> bool:
try:
umcph = self._umcphs.find_user_message_count_per_hour_by_user_id_and_date(user.id, created_at)
if umcph is None or umcph.xp_count is None:
return False
return umcph.xp_count > settings.max_message_xp_per_hour
return umcph.xp_count > settings.max_message_xp_per_hour
except Exception as e:
self._logger.error(
__name__,
f"Cannot add user message count per hour with",
e,
)
return False
def get_ontime_for_user(self, user: User) -> float:
return round(
self._user_joined_voice_channel.get_user_joined_voice_channels_by_user_id(user.id)
.where(lambda x: x.leaved_on is not None and x.joined_on is not None)
.sum(lambda join: (join.leaved_on - join.joined_on).total_seconds() / 3600),
sum(
[
(join.leaved_on - join.joined_on).total_seconds() / 3600
for join in self._user_joined_voice_channel.get_user_joined_voice_channels_by_user_id(user.id)
if join.leaved_on is not None and join.joined_on is not None
]
),
2,
)
@ -214,7 +219,7 @@ class ClientUtilsService(ClientUtilsABC):
guild: Guild = self._bot.guilds.where(lambda g: g == guild).single()
channel = guild.get_channel(discord_channel_id)
message = await channel.fetch_message(discord_message_id)
emoji = List(discord.Emoji, guild.emojis).where(lambda x: x.name == rule.emoji_name).single()
emoji = List(discord.Emoji, [x for x in guild.emojis if x.name == rule.emoji_name]).single()
if emoji is None:
self._logger.debug(__name__, f"Emoji {rule.emoji_name} not found")

546
bot/src/bot_core/service/data_integrity_service.py
View File

@ -1,9 +1,8 @@
from datetime import datetime, timedelta
from typing import Union
import discord
from cpl_core.configuration import ConfigurationABC
from cpl_core.database.context import DatabaseContextABC
from cpl_discord.container import Member, Guild
from cpl_discord.service import DiscordBotServiceABC
from bot_core.abc.client_utils_abc import ClientUtilsABC
@ -66,356 +65,255 @@ class DataIntegrityService:
self._is_for_shutdown = False
def _check_known_users(self):
self._logger.debug(__name__, f"Start checking KnownUsers table, {len(self._bot.users)}")
for u in self._bot.users:
u: discord.User = u
try:
if u.bot:
self._logger.trace(__name__, f"User {u.id} is ignored, because its a bot")
continue
async def check_data_integrity(self, is_for_shutdown=False):
self._logger.info(__name__, f"Data integrity service started")
if is_for_shutdown != self._is_for_shutdown:
self._is_for_shutdown = is_for_shutdown
user = self._known_users.find_user_by_discord_id(u.id)
if user is not None:
continue
try:
for g in self._bot.guilds:
self._logger.debug(__name__, f"Start check for server: {g.id}")
s = self._get_or_create_server(g)
self._logger.debug(__name__, f"Start check for clients")
self._check_clients(g.id, s)
self._logger.warn(__name__, f"Unknown user: {u.id}")
self._logger.debug(__name__, f"Add user: {u.id}")
self._known_users.add_user(KnownUser(u.id))
self._db_context.save_changes()
for m in [m for m in g.members if not m.bot]:
await self._check_default_role(m)
self._check_known_user(m.id)
user = self._known_users.find_user_by_discord_id(u.id)
if user is None:
self._logger.fatal(__name__, f"Cannot add user: {u.id}")
self._logger.debug(__name__, f"Start check for member: {g.id}@{m.id}")
u = self._get_or_create_user(s, m.id)
self._logger.debug(__name__, f"Added user: {u.id}")
except Exception as e:
self._logger.error(__name__, f"Cannot get user", e)
self._logger.debug(__name__, f"Start check for user joined server: {g.id}@{m.id}")
self._check_user_join(g, m, u)
def check_servers(self):
self._logger.debug(__name__, f"Start checking Servers table")
for g in self._bot.guilds:
g: discord.Guild = g
try:
server = self._servers.find_server_by_discord_id(g.id)
if server is not None:
continue
self._logger.debug(__name__, f"Start check for user joined voice channels: {g.id}@{m.id}")
self._check_user_joined_vc(g.id, m, u)
self._logger.warn(__name__, f"Server not found in database: {g.id}")
self._logger.debug(__name__, f"Add server: {g.id}")
self._servers.add_server(Server(g.id))
self._db_context.save_changes()
self._logger.debug(__name__, f"Start check for user joined game servers: {g.id}@{m.id}")
self._check_user_joined_gs(g.id, m.id, u)
server = self._servers.find_server_by_discord_id(g.id)
if server is None:
self._logger.fatal(__name__, f"Cannot add server: {g.id}")
self._logger.debug(__name__, f"Start check for user got achievements: {g.id}@{m.id}")
await self._check_for_user_achievements(u)
self._logger.debug(__name__, f"Added server: {g.id}")
except Exception as e:
self._logger.error(__name__, f"Cannot get server", e)
for m in [m for m in g.members if m.bot]:
u = self._users.find_user_by_discord_id_and_server_id(m.id, s.id)
if u is None:
continue
self._remove_bot(u)
self._logger.info(__name__, f"Data integrity service finished")
except Exception as e:
self._logger.fatal(__name__, f"Checking data integrity failed", e)
def _get_or_create_server(self, guild: Guild) -> Server:
try:
server = self._servers.find_server_by_discord_id(guild.id)
if server is not None:
return server
self._logger.warn(__name__, f"Server not found in database: {guild.id}")
self._logger.debug(__name__, f"Add server: {guild.id}")
self._servers.add_server(Server(guild.id))
self._db_context.save_changes()
server = self._servers.find_server_by_discord_id(guild.id)
if server is None:
self._logger.fatal(__name__, f"Cannot add server: {guild.id}")
self._logger.trace(__name__, f"Added server: {guild.id}")
return server
except Exception as e:
self._logger.error(__name__, f"Cannot get server", e)
def _check_clients(self, guild_id: int, server: Server):
try:
client = self._clients.find_client_by_server_id(server.id)
if client is not None:
return
self._logger.warn(
__name__,
f"Client for server {guild_id} not found in database: {self._bot.user.id}",
)
self._logger.debug(__name__, f"Add client: {self._bot.user.id}")
self._clients.add_client(Client(self._bot.user.id, 0, 0, 0, 0, 0, server))
self._db_context.save_changes()
client = self._clients.find_client_by_server_id(server.id)
if client is None:
self._logger.fatal(
__name__,
f"Cannot add client {self._bot.user.id} for server {guild_id}",
)
self._logger.trace(__name__, f"Added client: {guild_id}")
except Exception as e:
self._logger.error(__name__, f"Cannot get client", e)
results = self._servers.get_servers()
if results is None or len(results) == 0:
self._logger.error(__name__, f"Table Servers is empty!")
def _check_clients(self):
self._logger.debug(__name__, f"Start checking Clients table")
for g in self._bot.guilds:
g: discord.Guild = g
try:
server: Server = self._servers.find_server_by_discord_id(g.id)
if server is None:
self._logger.fatal(__name__, f"Server not found in database: {g.id}")
def _check_known_user(self, member_id: int):
try:
if self._known_users.find_user_by_discord_id(member_id) is not None:
return
client = self._clients.find_client_by_server_id(server.id)
if client is not None:
self._logger.warn(__name__, f"Unknown user: {member_id}")
self._logger.trace(__name__, f"Add known user: {member_id}")
self._known_users.add_user(KnownUser(member_id))
self._db_context.save_changes()
user = self._known_users.find_user_by_discord_id(member_id)
if user is None:
self._logger.fatal(__name__, f"Cannot add user: {member_id}")
self._logger.trace(__name__, f"Added known user: {member_id}")
except Exception as e:
self._logger.error(__name__, f"Cannot get user", e)
def _get_or_create_user(self, server: Server, member_id: int) -> User:
try:
user = self._users.find_user_by_discord_id_and_server_id(member_id, server.id)
if user is not None:
return user
self._logger.warn(__name__, f"User not found in database: {member_id}")
self._logger.debug(__name__, f"Add user: {member_id}")
self._users.add_user(User(member_id, 0, 0, 0, None, server))
self._db_context.save_changes()
self._logger.trace(__name__, f"Added User: {member_id}")
return self._users.get_user_by_discord_id_and_server_id(member_id, server.id)
except Exception as e:
self._logger.error(__name__, f"Cannot get User", e)
results = self._users.get_users()
if results is None or len(results) == 0:
self._logger.error(__name__, f"Table Users is empty!")
def _check_user_join(self, guild: Guild, member: Member, user: User):
try:
join = self._user_joins.find_active_user_joined_server_by_user_id(user.id)
if join is not None:
return
self._logger.warn(
__name__,
f"Active UserJoinedServer not found in database: {guild.id}:{member.id}@{member.joined_at}",
)
self._logger.debug(
__name__,
f"Add UserJoinedServer: {guild.id}:{member.id}@{member.joined_at}",
)
self._user_joins.add_user_joined_server(UserJoinedServer(user, self._dtp.transform(member.joined_at), None))
self._db_context.save_changes()
self._logger.trace(__name__, f"Added UserJoinedServer: {member.id}")
except Exception as e:
self._logger.error(__name__, f"Cannot get UserJoinedServer", e)
try:
joins = self._user_joins.get_user_joined_servers()
for join in [x for x in joins if x.user.server.discord_id == guild.id and x.leaved_on is None]:
dc_user = guild.get_member(join.user.discord_id)
if dc_user is not None:
continue
self._logger.warn(
__name__,
f"Client for server {g.id} not found in database: {self._bot.user.id}",
f"User {join.user.discord_id} already left the server.",
)
self._logger.debug(__name__, f"Add client: {self._bot.user.id}")
self._clients.add_client(Client(self._bot.user.id, 0, 0, 0, 0, 0, server))
join.leaved_on = datetime.now()
self._user_joins.update_user_joined_server(join)
self._db_context.save_changes()
except Exception as e:
self._logger.error(__name__, f"Cannot update UserJoinedServer", e)
client = self._clients.find_client_by_server_id(server.id)
if client is None:
self._logger.fatal(
__name__,
f"Cannot add client {self._bot.user.id} for server {g.id}",
)
def _check_user_joined_vc(self, guild_id: int, member: Member, user: User):
settings: ServerConfig = self._config.get_configuration(f"ServerConfig_{guild_id}")
self._logger.debug(__name__, f"Added client: {g.id}")
except Exception as e:
self._logger.error(__name__, f"Cannot get client", e)
try:
# close open voice states
joins = self._user_joins_vc.find_active_user_joined_voice_channels_by_user_id(user.id)
if joins is None or len(joins) == 0:
return
results = self._servers.get_servers()
if results is None or len(results) == 0:
self._logger.error(__name__, f"Table Servers is empty!")
def _check_users(self):
self._logger.debug(__name__, f"Start checking Users table")
for g in self._bot.guilds:
g: discord.Guild = g
try:
server = self._servers.find_server_by_discord_id(g.id)
if server is None:
self._logger.fatal(__name__, f"Server not found in database: {g.id}")
for u in g.members:
u: Union[discord.Member, discord.User] = u
if u.bot:
self._logger.trace(__name__, f"User {u.id} is ignored, because its a bot")
continue
user = self._users.find_user_by_discord_id_and_server_id(u.id, server.id)
if user is not None:
continue
self._logger.warn(__name__, f"User not found in database: {u.id}")
self._logger.debug(__name__, f"Add user: {u.id}")
self._users.add_user(User(u.id, 0, 0, 0, None, server))
self._db_context.save_changes()
self._logger.debug(__name__, f"Added User: {u.id}")
except Exception as e:
self._logger.error(__name__, f"Cannot get User", e)
results = self._users.get_users()
if results is None or len(results) == 0:
self._logger.error(__name__, f"Table Users is empty!")
def _check_user_joins(self):
self._logger.debug(__name__, f"Start checking UserJoinedServers table")
for guild in self._bot.guilds:
guild: discord.Guild = guild
server = self._servers.find_server_by_discord_id(guild.id)
if server is None:
self._logger.fatal(__name__, f"Server not found in database: {guild.id}")
try:
for u in guild.members:
u: discord.User = u
if u.bot:
self._logger.trace(__name__, f"User {u.id} is ignored, because its a bot")
continue
user = self._users.find_user_by_discord_id_and_server_id(u.id, server.id)
if user is None:
self._logger.fatal(__name__, f"User not found in database: {u.id}")
join = self._user_joins.find_active_user_joined_server_by_user_id(user.id)
if join is not None:
continue
m: discord.Member = u
self._logger.warn(
__name__,
f"Active UserJoinedServer not found in database: {guild.id}:{u.id}@{m.joined_at}",
)
self._logger.debug(
__name__,
f"Add UserJoinedServer: {guild.id}:{u.id}@{m.joined_at}",
)
self._user_joins.add_user_joined_server(
UserJoinedServer(user, self._dtp.transform(m.joined_at), None)
)
self._db_context.save_changes()
self._logger.debug(__name__, f"Added UserJoinedServer: {u.id}")
except Exception as e:
self._logger.error(__name__, f"Cannot get UserJoinedServer", e)
results = self._users.get_users()
if results is None or len(results) == 0:
self._logger.error(__name__, f"Table Users is empty!")
joins = self._user_joins.get_user_joined_servers()
for join in joins:
join: UserJoinedServer = join
if join.user.server.discord_id != guild.id:
continue
self._logger.warn(
__name__,
f"Active UserJoinedVoiceChannel found in database: {guild_id}:{member.id}@{join.joined_on}",
)
join.leaved_on = datetime.now()
if join.leaved_on is not None:
continue
if ((join.leaved_on - join.joined_on).total_seconds() / 60 / 60) > settings.max_voice_state_hours:
join.leaved_on = join.joined_on + timedelta(hours=settings.max_voice_state_hours)
dc_user = guild.get_member(join.user.discord_id)
if dc_user is None:
self._logger.warn(
__name__,
f"User {join.user.discord_id} already left the server.",
)
join.leaved_on = datetime.now()
self._user_joins.update_user_joined_server(join)
self._user_joins_vc.update_user_joined_voice_channel(join)
self._db_context.save_changes()
if self._is_for_shutdown:
user.xp += round(join.time * settings.xp_per_ontime_hour)
self._users.update_user(user)
def _check_user_joins_vc(self):
self._logger.debug(__name__, f"Start checking UserJoinedVoiceChannel table")
for guild in self._bot.guilds:
guild: discord.Guild = guild
settings: ServerConfig = self._config.get_configuration(f"ServerConfig_{guild.id}")
server = self._servers.find_server_by_discord_id(guild.id)
if server is None:
self._logger.fatal(__name__, f"Server not found in database: {guild.id}")
try:
# close open voice states
for member in guild.members:
if member.bot:
self._logger.trace(__name__, f"User {member.id} is ignored, because its a bot")
continue
user = self._users.find_user_by_discord_id_and_server_id(member.id, server.id)
if user is None:
self._logger.fatal(__name__, f"User not found in database: {member.id}")
joins = self._user_joins_vc.find_active_user_joined_voice_channels_by_user_id(user.id)
if joins is None or len(joins) == 0:
continue
for join in joins:
self._logger.warn(
__name__,
f"Active UserJoinedVoiceChannel found in database: {guild.id}:{member.id}@{join.joined_on}",
)
join.leaved_on = datetime.now()
if (
(join.leaved_on - join.joined_on).total_seconds() / 60 / 60
) > settings.max_voice_state_hours:
join.leaved_on = join.joined_on + timedelta(hours=settings.max_voice_state_hours)
self._user_joins_vc.update_user_joined_voice_channel(join)
if self._is_for_shutdown:
user.xp += round(join.time * settings.xp_per_ontime_hour)
self._users.update_user(user)
self._db_context.save_changes()
if self._is_for_shutdown:
return
# add open voice states
for member in guild.members:
if member.bot:
self._logger.trace(__name__, f"User {member.id} is ignored, because its a bot")
continue
if member.voice is None or member.voice.channel.id in settings.afk_channel_ids:
continue
user = self._users.find_user_by_discord_id_and_server_id(member.id, server.id)
if user is None:
self._logger.fatal(__name__, f"User not found in database: {member.id}")
join = UserJoinedVoiceChannel(user, member.voice.channel.id, datetime.now())
self._user_joins_vc.add_user_joined_voice_channel(join)
self._db_context.save_changes()
except Exception as e:
self._logger.error(__name__, f"Cannot get UserJoinedVoiceChannel", e)
def _check_user_joined_gs(self):
self._logger.debug(__name__, f"Start checking UserJoinedGameServer table")
for guild in self._bot.guilds:
guild: discord.Guild = guild
server = self._servers.find_server_by_discord_id(guild.id)
if server is None:
self._logger.fatal(__name__, f"Server not found in database: {guild.id}")
try:
for member in guild.members:
if member.bot:
self._logger.trace(__name__, f"User {member.id} is ignored, because its a bot")
continue
user = self._users.find_user_by_discord_id_and_server_id(member.id, server.id)
if user is None:
self._logger.fatal(__name__, f"User not found in database: {member.id}")
joins = self._user_joined_gs.find_active_user_joined_game_servers_by_user_id(user.id)
if joins is None or len(joins) == 0:
continue
for join in joins:
self._logger.warn(
__name__,
f"Active UserJoinedGameServer found in database: {guild.id}:{member.id}@{join.joined_on}",
)
join.leaved_on = datetime.now()
settings: ServerConfig = self._config.get_configuration(f"ServerConfig_{guild.id}")
if (
(join.leaved_on - join.joined_on).total_seconds() / 60 / 60
) > settings.max_voice_state_hours:
join.leaved_on = join.joined_on + timedelta(hours=settings.max_voice_state_hours)
self._user_joined_gs.update_user_joined_game_server(join)
if self._is_for_shutdown:
user.xp += round(join.time * settings.xp_per_ontime_hour)
self._users.update_user(user)
self._db_context.save_changes()
except Exception as e:
self._logger.error(__name__, f"Cannot get UserJoinedGameServer", e)
async def _check_for_user_achievements(self):
self._logger.debug(__name__, f"Start checking UserGotAchievement table")
for guild in self._bot.guilds:
server = self._servers.find_server_by_discord_id(guild.id)
if server is None:
self._logger.fatal(__name__, f"Server not found in database: {guild.id}")
for member in guild.members:
if member.bot:
self._logger.trace(__name__, f"User {member.id} is ignored, because its a bot")
continue
user = self._users.find_user_by_discord_id_and_server_id(member.id, server.id)
if user is None:
self._logger.fatal(__name__, f"User not found in database: {member.id}")
await self._achievements.validate_achievements_for_user(user)
async def _check_default_role(self):
for guild in self._bot.guilds:
for member in guild.members:
await self._client_utils.check_default_role(member)
def _check_for_bots(self):
for guild in self._bot.guilds:
server = self._servers.get_server_by_discord_id(guild.id)
for member in guild.members.where(lambda x: x.bot):
user = self._users.find_user_by_discord_id_and_server_id(member.id, server.id)
if user is None:
continue
for join in self._user_joins.get_user_joined_servers_by_user_id(user.id):
self._user_joins.delete_user_joined_server(join)
self._user_joins_vc.delete_user_joined_voice_channel_by_user_id(user.id)
self._users.delete_user(user)
self._db_context.save_changes()
if self._is_for_shutdown:
return
async def check_data_integrity(self, is_for_shutdown=False):
if is_for_shutdown != self._is_for_shutdown:
self._is_for_shutdown = is_for_shutdown
# add open voice states
if member.voice is None or member.voice.channel.id in settings.afk_channel_ids:
return
await self._check_default_role()
self._check_known_users()
self.check_servers()
self._check_clients()
self._check_users()
self._check_user_joins()
self._check_user_joins_vc()
self._check_user_joined_gs()
await self._check_for_user_achievements()
self._check_for_bots()
join = UserJoinedVoiceChannel(user, member.voice.channel.id, datetime.now())
self._user_joins_vc.add_user_joined_voice_channel(join)
self._db_context.save_changes()
except Exception as e:
self._logger.error(__name__, f"Cannot get UserJoinedVoiceChannel", e)
def _check_user_joined_gs(self, guild_id: int, member_id: int, user: User):
try:
joins = self._user_joined_gs.find_active_user_joined_game_servers_by_user_id(user.id)
if joins is None or len(joins) == 0:
return
for join in joins:
self._logger.warn(
__name__,
f"Active UserJoinedGameServer found in database: {guild_id}:{member_id}@{join.joined_on}",
)
join.leaved_on = datetime.now()
settings: ServerConfig = self._config.get_configuration(f"ServerConfig_{guild_id}")
if join.time > settings.max_voice_state_hours:
join.leaved_on = join.joined_on + timedelta(hours=settings.max_voice_state_hours)
self._user_joined_gs.update_user_joined_game_server(join)
if self._is_for_shutdown:
user.xp += round(join.time * settings.xp_per_ontime_hour)
self._users.update_user(user)
self._db_context.save_changes()
except Exception as e:
self._logger.error(__name__, f"Cannot get UserJoinedGameServer", e)
async def _check_for_user_achievements(self, user: User):
try:
await self._achievements.validate_achievements_for_user(user)
except Exception as e:
self._logger.error(__name__, f"Cannot check UserGotAchievement for {user.id}", e)
async def _check_default_role(self, member: Member):
await self._client_utils.check_default_role(member)
def _remove_bot(self, user: User):
known_user = self._known_users.find_user_by_discord_id(user.discord_id)
if known_user is not None:
self._known_users.delete_user(known_user)
for join in self._user_joins.get_user_joined_servers_by_user_id(user.id):
self._user_joins.delete_user_joined_server(join)
self._user_joins_vc.delete_user_joined_voice_channel_by_user_id(user.id)
self._users.delete_user(user)
self._db_context.save_changes()

114
bot/src/bot_core/service/permission_service.py Normal file
View File

@ -0,0 +1,114 @@
from typing import Optional
import discord
from cpl_core.configuration import ConfigurationABC
from cpl_core.logging import LoggerABC
from cpl_discord.service import DiscordBotServiceABC
from bot_data.abc.server_config_repository_abc import ServerConfigRepositoryABC
from bot_data.abc.server_repository_abc import ServerRepositoryABC
from bot_data.abc.technician_config_repository_abc import TechnicianConfigRepositoryABC
from bot_data.model.team_member_type_enum import TeamMemberTypeEnum
from bot_core.abc.permission_service_abc import PermissionServiceABC
class PermissionService(PermissionServiceABC):
def __init__(
self,
logger: LoggerABC,
bot: DiscordBotServiceABC,
config: ConfigurationABC,
servers: ServerRepositoryABC,
server_configs: ServerConfigRepositoryABC,
technician_configs: TechnicianConfigRepositoryABC,
):
PermissionServiceABC.__init__(self)
self._logger = logger
self._bot = bot
self._config = config
self._servers = servers
self._server_configs = server_configs
self._technician_configs = technician_configs
# member_id: {team_member_type: {guild_id: bool}}
self._cache: dict[int, dict[TeamMemberTypeEnum, dict[int, bool]]] = {}
def reset_cache(self):
self._cache = {}
def get_cached_permission(
self, member_id: int, team_member_type: TeamMemberTypeEnum, guild_id: int = None
) -> Optional[bool]:
if member_id not in self._cache:
self._cache[member_id] = {}
if team_member_type not in self._cache[member_id]:
self._cache[member_id][team_member_type] = {}
return None
if guild_id not in self._cache[member_id][team_member_type]:
return None
return self._cache[member_id][team_member_type][guild_id]
def set_cached_permission(
self, value: bool, member_id: int, team_member_type: TeamMemberTypeEnum, guild_id: int = None
):
if member_id not in self._cache:
self._cache[member_id] = {}
if team_member_type not in self._cache[member_id]:
self._cache[member_id][team_member_type] = {}
self._cache[member_id][team_member_type][guild_id] = value
def _has_member_role(self, member: discord.Member, team_member_type: TeamMemberTypeEnum) -> bool:
if member is None or member.guild is None:
return False
try:
has_permission_cached = self.get_cached_permission(member.id, team_member_type, member.guild.id)
if has_permission_cached is not None:
return has_permission_cached
self._logger.debug(__name__, f"Checking is member {member.name} {team_member_type.value}")
has_permission = True in [
member.guild.get_role(x.role_id) not in member.roles
for x in self._server_configs.get_server_config_by_server(
self._servers.get_server_by_discord_id(member.guild.id).id
).team_role_ids
if x.team_member_type == team_member_type
]
self.set_cached_permission(has_permission, member.id, team_member_type, member.guild.id)
return has_permission
except Exception as e:
self._logger.error(__name__, "Permission check failed", e)
return False
def is_member_admin(self, member: discord.Member) -> bool:
return self._has_member_role(member, TeamMemberTypeEnum.admin)
def is_member_moderator(self, member: discord.Member) -> bool:
return self._has_member_role(member, TeamMemberTypeEnum.moderator) or self._has_member_role(
member, TeamMemberTypeEnum.admin
)
def is_member_technician(self, member: discord.Member) -> bool:
if member is None or member.guild is None:
return False
has_permission_cached = self.get_cached_permission(member.id, TeamMemberTypeEnum.technician)
if has_permission_cached is not None:
return has_permission_cached
self._logger.debug(__name__, f"Checking is member {member.name} technician")
try:
has_permission = member.id in self._technician_configs.get_technician_config().technician_ids
self.set_cached_permission(has_permission, member.id, TeamMemberTypeEnum.technician)
return has_permission
except Exception as e:
self._logger.error(__name__, "Permission check failed", e)
return False

4
bot/src/bot_data/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_data"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

4
bot/src/bot_data/abc/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_data.abc"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

4
bot/src/bot_data/abc/user_joined_server_repository_abc.py
View File

@ -19,10 +19,6 @@ class UserJoinedServerRepositoryABC(ABC):
def get_user_joined_server_by_id(self, id: int) -> UserJoinedServer:
pass
@abstractmethod
def get_user_joined_server_by_server_id(self, server_id: int) -> UserJoinedServer:
pass
@abstractmethod
def get_user_joined_servers_by_user_id(self, user_id: int) -> list[UserJoinedServer]:
pass

2
bot/src/bot_data/bot-data.json
View File

@ -4,7 +4,7 @@
"Version": {
"Major": "1",
"Minor": "2",
"Micro": "2"
"Micro": "3"
},
"Author": "Sven Heidemann",
"AuthorEmail": "sven.heidemann@sh-edraft.de",

4
bot/src/bot_data/model/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_data.model"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

1
bot/src/bot_data/model/team_member_type_enum.py
View File

@ -4,3 +4,4 @@ from enum import Enum
class TeamMemberTypeEnum(Enum):
moderator = "Moderator"
admin = "Admin"
technician = "Technician"

29
bot/src/bot_data/model/user.py
View File

@ -2,6 +2,7 @@ from datetime import datetime, date
from typing import Optional
from cpl_core.database import TableABC
from cpl_core.database.context import DatabaseContextABC
from cpl_core.dependency_injection import ServiceProviderABC
from cpl_discord.service import DiscordBotServiceABC
from cpl_query.extension import List
@ -92,10 +93,32 @@ class User(TableABC):
@property
@ServiceProviderABC.inject
def ontime(self, services: ServiceProviderABC) -> float:
from bot_core.abc.client_utils_abc import ClientUtilsABC
db: DatabaseContextABC = services.get_service(DatabaseContextABC)
result = db.select(
f"""
SELECT ROUND(SUM(TIME_TO_SEC(TIMEDIFF(UserJoinedVoiceChannel.LeavedOn, UserJoinedVoiceChannel.JoinedOn)) / 3600),2)
FROM UserJoinedVoiceChannel
WHERE UserId = {self._user_id};
"""
)[0][0]
if result is None:
return 0
return float(result)
client_utils: ClientUtilsABC = services.get_service(ClientUtilsABC)
return client_utils.get_ontime_for_user(self)
@property
@ServiceProviderABC.inject
def game_ontime(self, services: ServiceProviderABC) -> float:
db: DatabaseContextABC = services.get_service(DatabaseContextABC)
result = db.select(
f"""
SELECT ROUND(SUM(TIME_TO_SEC(TIMEDIFF(UserJoinedGameServer.LeavedOn, UserJoinedGameServer.JoinedOn)) / 3600),2)
FROM UserJoinedGameServer
WHERE UserId = {self._user_id};
"""
)[0][0]
if result is None:
return 0
return float(result)
@property
@ServiceProviderABC.inject

9
bot/src/bot_data/model/user_joined_server.py
View File

@ -67,15 +67,6 @@ class UserJoinedServer(TableABC):
"""
)
@staticmethod
def get_select_by_server_id_string(id: int) -> str:
return str(
f"""
SELECT * FROM `UserJoinedServers`
WHERE `ServerId` = {id};
"""
)
@staticmethod
def get_select_by_user_id_string(id: int) -> str:
return str(

4
bot/src/bot_data/service/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_data.service"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

62
bot/src/bot_data/service/achievements_repository_service.py
View File

@ -48,66 +48,60 @@ class AchievementRepositoryService(AchievementRepositoryABC):
)
def get_achievements(self) -> List[Achievement]:
achievements = List(Achievement)
self._logger.trace(__name__, f"Send SQL command: {Achievement.get_select_all_string()}")
results = self._context.select(Achievement.get_select_all_string())
for result in results:
self._logger.trace(__name__, f"Get user with id {result[0]}")
achievements.append(self._from_result(result))
return achievements
return List(
Achievement,
[self._from_result(result) for result in self._context.select(Achievement.get_select_all_string())],
)
def get_achievement_by_id(self, id: int) -> Achievement:
self._logger.trace(__name__, f"Send SQL command: {Achievement.get_select_by_id_string(id)}")
result = self._context.select(Achievement.get_select_by_id_string(id))[0]
return self._from_result(result)
return self._from_result(self._context.select(Achievement.get_select_by_id_string(id))[0])
def get_achievements_by_server_id(self, server_id: int) -> List[Achievement]:
achievements = List(Achievement)
self._logger.trace(
__name__,
f"Send SQL command: {Achievement.get_select_by_server_id_string(server_id)}",
)
results = self._context.select(Achievement.get_select_by_server_id_string(server_id))
for result in results:
self._logger.trace(__name__, f"Get user with id {result[0]}")
achievements.append(self._from_result(result))
return achievements
return List(
Achievement,
[
self._from_result(result)
for result in self._context.select(Achievement.get_select_by_server_id_string(server_id))
],
)
def get_achievements_by_user_id(self, user_id: int) -> List[Achievement]:
achievements = List(Achievement)
achievements_joins = List(UserGotAchievement)
self._logger.trace(
__name__,
f"Send SQL command: {UserGotAchievement.get_select_by_user_id_string(user_id)}",
)
results = self._context.select(UserGotAchievement.get_select_by_user_id_string(user_id))
for result in results:
self._logger.trace(__name__, f"Got UserGotAchievement with id {result[0]}")
achievements_joins.append(self._join_from_result(result))
for achievements_join in achievements_joins:
results = self._context.select(Achievement.get_select_by_id_string(achievements_join.achievement.id))
for result in results:
self._logger.trace(__name__, f"Got Achievement with id {result[0]}")
achievements.append(self._from_result(result))
return achievements
return List(
UserGotAchievement,
[
self._join_from_result(result).achievement
for result in self._context.select(UserGotAchievement.get_select_by_user_id_string(user_id))
],
)
def get_user_got_achievements_by_achievement_id(self, achievement_id: int) -> List[Achievement]:
achievements_joins = List(UserGotAchievement)
self._logger.trace(
__name__,
f"Send SQL command: {UserGotAchievement.get_select_by_achievement_id_string(achievement_id)}",
)
results = self._context.select(UserGotAchievement.get_select_by_achievement_id_string(achievement_id))
for result in results:
self._logger.trace(__name__, f"Got UserGotAchievement with id {result[0]}")
achievements_joins.append(self._join_from_result(result))
return achievements_joins
return List(
UserGotAchievement,
[
self._join_from_result(result)
for result in self._context.select(
UserGotAchievement.get_select_by_achievement_id_string(achievement_id)
)
],
)
def add_achievement(self, achievement: Achievement):
self._logger.trace(__name__, f"Send SQL command: {achievement.insert_string}")

9
bot/src/bot_data/service/api_key_repository_service.py
View File

@ -44,13 +44,12 @@ class ApiKeyRepositoryService(ApiKeyRepositoryABC):
return api_key
def get_api_keys(self) -> List[ApiKey]:
api_keys = List(ApiKey)
self._logger.trace(__name__, f"Send SQL command: {ApiKey.get_select_all_string()}")
results = self._context.select(ApiKey.get_select_all_string())
for result in results:
api_keys.append(self._api_key_from_result(result))
return api_keys
return List(
ApiKey,
[self._api_key_from_result(result) for result in self._context.select(ApiKey.get_select_all_string())],
)
def get_api_key(self, identifier: str, key: str) -> ApiKey:
self._logger.trace(__name__, f"Send SQL command: {ApiKey.get_select_string(identifier, key)}")

32
bot/src/bot_data/service/auth_user_repository_service.py
View File

@ -64,23 +64,16 @@ class AuthUserRepositoryService(AuthUserRepositoryABC):
__name__,
f"Send SQL command: {auth_user.get_select_user_id_from_relations()}",
)
relation_ids = List(int)
results = self._context.select(auth_user.get_select_user_id_from_relations())
for result in results:
self._logger.trace(__name__, f"Got auth user relation with id {result[0]}")
relation_ids.append(result[0])
return relation_ids
return List(int, [result[0] for result in self._context.select(auth_user.get_select_user_id_from_relations())])
def get_all_auth_users(self) -> List[AuthUser]:
users = List(AuthUser)
self._logger.trace(__name__, f"Send SQL command: {AuthUser.get_select_all_string()}")
results = self._context.select(AuthUser.get_select_all_string())
for result in results:
self._logger.trace(__name__, f"Get auth user with id {result[0]}")
users.append(self._user_from_result(result))
return users
return List(
AuthUser,
[self._user_from_result(result) for result in self._context.select(AuthUser.get_select_all_string())],
)
def get_filtered_auth_users(self, criteria: AuthUserSelectCriteria) -> FilteredResult:
users = self.get_all_auth_users()
@ -129,8 +122,7 @@ class AuthUserRepositoryService(AuthUserRepositoryABC):
def get_auth_user_by_email(self, email: str) -> AuthUser:
self._logger.trace(__name__, f"Send SQL command: {AuthUser.get_select_by_email_string(email)}")
result = self._context.select(AuthUser.get_select_by_email_string(email))[0]
return self._user_from_result(result)
return self._user_from_result(self._context.select(AuthUser.get_select_by_email_string(email))[0])
def find_auth_user_by_email(self, email: str) -> Optional[AuthUser]:
self._logger.trace(__name__, f"Send SQL command: {AuthUser.get_select_by_email_string(email)}")
@ -138,9 +130,7 @@ class AuthUserRepositoryService(AuthUserRepositoryABC):
if result is None or len(result) == 0:
return None
result = result[0]
return self._user_from_result(result)
return self._user_from_result(result[0])
def find_auth_user_by_confirmation_id(self, id: str) -> Optional[AuthUser]:
self._logger.trace(
@ -151,9 +141,7 @@ class AuthUserRepositoryService(AuthUserRepositoryABC):
if result is None or len(result) == 0:
return None
result = result[0]
return self._user_from_result(result)
return self._user_from_result(result[0])
def find_auth_user_by_forgot_password_id(self, id: str) -> Optional[AuthUser]:
self._logger.trace(
@ -164,9 +152,7 @@ class AuthUserRepositoryService(AuthUserRepositoryABC):
if result is None or len(result) == 0:
return None
result = result[0]
return self._user_from_result(result)
return self._user_from_result(result[0])
def add_auth_user(self, user: AuthUser):
self._logger.trace(__name__, f"Send SQL command: {user.insert_string}")

138
bot/src/bot_data/service/auto_role_repository_service.py
View File

@ -23,27 +23,7 @@ class AutoRoleRepositoryService(AutoRoleRepositoryABC):
AutoRoleRepositoryABC.__init__(self)
def get_auto_roles(self) -> List[AutoRole]:
auto_roles = List(AutoRole)
self._logger.trace(__name__, f"Send SQL command: {AutoRole.get_select_all_string()}")
results = self._context.select(AutoRole.get_select_all_string())
for result in results:
auto_roles.append(
AutoRole(
self._servers.get_server_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
)
return auto_roles
def get_auto_role_by_id(self, id: int) -> AutoRole:
self._logger.trace(__name__, f"Send SQL command: {AutoRole.get_select_by_id_string(id)}")
result = self._context.select(AutoRole.get_select_by_id_string(id))[0]
def _from_result(self, result: tuple) -> AutoRole:
return AutoRole(
self._servers.get_server_by_id(result[1]),
result[2],
@ -53,55 +33,37 @@ class AutoRoleRepositoryService(AutoRoleRepositoryABC):
id=result[0],
)
def get_auto_roles(self) -> List[AutoRole]:
self._logger.trace(__name__, f"Send SQL command: {AutoRole.get_select_all_string()}")
return List(
AutoRole, [self._from_result(result) for result in self._context.select(AutoRole.get_select_all_string())]
)
def get_auto_role_by_id(self, id: int) -> AutoRole:
self._logger.trace(__name__, f"Send SQL command: {AutoRole.get_select_by_id_string(id)}")
return self._from_result(self._context.select(AutoRole.get_select_by_id_string(id))[0])
def find_auto_role_by_id(self, id: int) -> Optional[AutoRole]:
self._logger.trace(__name__, f"Send SQL command: {AutoRole.get_select_by_id_string(id)}")
result = self._context.select(AutoRole.get_select_by_id_string(id))
if result is None or len(result) == 0:
return None
result = result[0]
return AutoRole(
self._servers.get_server_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
return self._from_result(result[0])
def get_auto_roles_by_server_id(self, id: int) -> List[AutoRole]:
self._logger.trace(__name__, f"Send SQL command: {AutoRole.get_select_by_server_id_string(id)}")
auto_roles = List(AutoRole)
results = self._context.select(AutoRole.get_select_by_server_id_string(id))
for result in results:
auto_roles.append(
AutoRole(
self._servers.get_server_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
)
return auto_roles
return List(
AutoRole,
[self._from_result(result) for result in self._context.select(AutoRole.get_select_by_server_id_string(id))],
)
def get_auto_role_by_message_id(self, id: int) -> AutoRole:
self._logger.trace(
__name__,
f"Send SQL command: {AutoRole.get_select_by_message_id_string(id)}",
)
result = self._context.select(AutoRole.get_select_by_message_id_string(id))[0]
return AutoRole(
self._servers.get_server_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
return self._from_result(self._context.select(AutoRole.get_select_by_message_id_string(id))[0])
def find_auto_role_by_message_id(self, id: int) -> Optional[AutoRole]:
self._logger.trace(
@ -112,16 +74,7 @@ class AutoRoleRepositoryService(AutoRoleRepositoryABC):
if result is None or len(result) == 0:
return None
result = result[0]
return AutoRole(
self._servers.get_server_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
return self._from_result(result[0])
def add_auto_role(self, auto_role: AutoRole):
self._logger.trace(__name__, f"Send SQL command: {auto_role.insert_string}")
@ -135,27 +88,7 @@ class AutoRoleRepositoryService(AutoRoleRepositoryABC):
self._logger.trace(__name__, f"Send SQL command: {auto_role.delete_string}")
self._context.cursor.execute(auto_role.delete_string)
def get_auto_role_rules(self) -> List[AutoRoleRule]:
auto_role_rules = List(AutoRoleRule)
self._logger.trace(__name__, f"Send SQL command: {AutoRoleRule.get_select_all_string()}")
results = self._context.select(AutoRoleRule.get_select_all_string())
for result in results:
auto_role_rules.append(
AutoRoleRule(
self.get_auto_role_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
)
return auto_role_rules
def get_auto_role_rule_by_id(self, id: int) -> AutoRoleRule:
self._logger.trace(__name__, f"Send SQL command: {AutoRoleRule.get_select_by_id_string(id)}")
result = self._context.select(AutoRoleRule.get_select_by_id_string(id))[0]
def _rule_from_result(self, result: tuple) -> AutoRoleRule:
return AutoRoleRule(
self.get_auto_role_by_id(result[1]),
result[2],
@ -165,26 +98,29 @@ class AutoRoleRepositoryService(AutoRoleRepositoryABC):
id=result[0],
)
def get_auto_role_rules(self) -> List[AutoRoleRule]:
self._logger.trace(__name__, f"Send SQL command: {AutoRoleRule.get_select_all_string()}")
return List(
AutoRoleRule,
[self._rule_from_result(result) for result in self._context.select(AutoRoleRule.get_select_all_string())],
)
def get_auto_role_rule_by_id(self, id: int) -> AutoRoleRule:
self._logger.trace(__name__, f"Send SQL command: {AutoRoleRule.get_select_by_id_string(id)}")
return self._rule_from_result(self._context.select(AutoRoleRule.get_select_by_id_string(id))[0])
def get_auto_role_rules_by_auto_role_id(self, id: int) -> List[AutoRoleRule]:
auto_role_rules = List(AutoRoleRule)
self._logger.trace(
__name__,
f"Send SQL command: {AutoRoleRule.get_select_by_auto_role_id_string(id)}",
)
results = self._context.select(AutoRoleRule.get_select_by_auto_role_id_string(id))
for result in results:
auto_role_rules.append(
AutoRoleRule(
self.get_auto_role_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
)
return auto_role_rules
return List(
AutoRoleRule,
[
self._rule_from_result(result)
for result in self._context.select(AutoRoleRule.get_select_by_auto_role_id_string(id))
],
)
def add_auto_role_rule(self, auto_role_rule: AutoRoleRule):
self._logger.trace(__name__, f"Send SQL command: {auto_role_rule.insert_string}")

122
bot/src/bot_data/service/client_repository_service.py
View File

@ -23,32 +23,7 @@ class ClientRepositoryService(ClientRepositoryABC):
ClientRepositoryABC.__init__(self)
def get_clients(self) -> List[Client]:
clients = List(Client)
self._logger.trace(__name__, f"Send SQL command: {Client.get_select_all_string()}")
results = self._context.select(Client.get_select_all_string())
for result in results:
self._logger.trace(__name__, f"Get client with id {result[0]}")
clients.append(
Client(
result[1],
result[2],
result[3],
result[4],
result[5],
result[6],
self._servers.get_server_by_id(result[7]),
result[8],
result[9],
id=result[0],
)
)
return clients
def get_client_by_id(self, client_id: int) -> Client:
self._logger.trace(__name__, f"Send SQL command: {Client.get_select_by_id_string(client_id)}")
result = self._context.select(Client.get_select_by_id_string(client_id))
def _from_result(self, result: tuple) -> Client:
return Client(
result[1],
result[2],
@ -62,49 +37,37 @@ class ClientRepositoryService(ClientRepositoryABC):
id=result[0],
)
def get_clients(self) -> List[Client]:
self._logger.trace(__name__, f"Send SQL command: {Client.get_select_all_string()}")
return List(
Client, [self._from_result(result) for result in self._context.select(Client.get_select_all_string())]
)
def get_client_by_id(self, client_id: int) -> Client:
self._logger.trace(__name__, f"Send SQL command: {Client.get_select_by_id_string(client_id)}")
return self._from_result(self._context.select(Client.get_select_by_id_string(client_id))[0])
def get_clients_by_server_id(self, server_id: int) -> List[Client]:
clients = List(Client)
self._logger.trace(
__name__,
f"Send SQL command: {Client.get_select_by_server_id_string(server_id)}",
)
results = self._context.select(Client.get_select_by_server_id_string(server_id))
for result in results:
clients.append(
Client(
result[1],
result[2],
result[3],
result[4],
result[5],
result[6],
self._servers.get_server_by_id(result[7]),
result[8],
result[9],
id=result[0],
)
)
return clients
return List(
Client,
[
self._from_result(result)
for result in self._context.select(Client.get_select_by_server_id_string(server_id))
],
)
def get_client_by_discord_id(self, discord_id: int) -> Client:
self._logger.trace(
__name__,
f"Send SQL command: {Client.get_select_by_discord_id_string(discord_id)}",
)
result = self._context.select(Client.get_select_by_discord_id_string(discord_id))[0]
return Client(
result[1],
result[2],
result[3],
result[4],
result[5],
result[6],
self._servers.get_server_by_id(result[7]),
result[8],
result[9],
id=result[0],
)
return self._from_result(self._context.select(Client.get_select_by_discord_id_string(discord_id))[0])
def find_client_by_discord_id(self, discord_id: int) -> Optional[Client]:
self._logger.trace(
@ -115,20 +78,7 @@ class ClientRepositoryService(ClientRepositoryABC):
if result is None or len(result) == 0:
return None
result = result[0]
return Client(
result[1],
result[2],
result[3],
result[4],
result[5],
result[6],
self._servers.get_server_by_id(result[7]),
result[8],
result[9],
id=result[0],
)
return self._from_result(result[0])
def find_client_by_server_id(self, discord_id: int) -> Optional[Client]:
self._logger.trace(
@ -139,20 +89,7 @@ class ClientRepositoryService(ClientRepositoryABC):
if result is None or len(result) == 0:
return None
result = result[0]
return Client(
result[1],
result[2],
result[3],
result[4],
result[5],
result[6],
self._servers.get_server_by_id(result[7]),
result[8],
result[9],
id=result[0],
)
return self._from_result(result[0])
def find_client_by_discord_id_and_server_id(self, discord_id: int, server_id: int) -> Optional[Client]:
self._logger.trace(
@ -163,20 +100,7 @@ class ClientRepositoryService(ClientRepositoryABC):
if result is None or len(result) == 0:
return None
result = result[0]
return Client(
result[1],
result[2],
result[3],
result[4],
result[5],
result[6],
self._servers.get_server_by_id(result[7]),
result[8],
result[9],
id=result[0],
)
return self._from_result(result[0])
def add_client(self, client: Client):
self._logger.trace(__name__, f"Send SQL command: {client.insert_string}")

38
bot/src/bot_data/service/game_server_repository_service.py
View File

@ -35,49 +35,49 @@ class GameServerRepositoryService(GameServerRepositoryABC):
)
def get_game_servers(self) -> List[GameServer]:
game_servers = List(GameServer)
self._logger.trace(
__name__,
f"Send SQL command: {GameServer.get_select_all_string()}",
)
results = self._context.select(GameServer.get_select_all_string())
for result in results:
self._logger.trace(__name__, f"Get user-joined-game-server with id {result[0]}")
game_servers.append(self._from_result(result))
return game_servers
return List(
GameServer,
[self._from_result(result) for result in self._context.select(GameServer.get_select_all_string())],
)
def get_game_servers_by_server_id(self, id: int) -> List[GameServer]:
game_servers = List(GameServer)
self._logger.trace(
__name__,
f"Send SQL command: {GameServer.get_select_by_server_id_string(id)}",
)
results = self._context.select(GameServer.get_select_by_server_id_string(id))
for result in results:
self._logger.trace(__name__, f"Get user-joined-game-server with id {result[0]}")
game_servers.append(self._from_result(result))
return game_servers
return List(
GameServer,
[
self._from_result(result)
for result in self._context.select(GameServer.get_select_by_server_id_string(id))
],
)
def get_game_server_by_id(self, id: int) -> GameServer:
self._logger.trace(
__name__,
f"Send SQL command: {GameServer.get_select_by_id_string(id)}",
)
result = self._context.select(GameServer.get_select_by_id_string(id))[0]
return self._from_result(result)
return self._from_result(self._context.select(GameServer.get_select_by_id_string(id))[0])
def get_game_servers_by_api_key_id(self, id: int) -> List[GameServer]:
self._logger.trace(
__name__,
f"Send SQL command: {GameServer.get_select_by_api_key_id_string(id)}",
)
game_servers = List(GameServer)
results = self._context.select(GameServer.get_select_by_api_key_id_string(id))
for result in results:
game_servers.append(self._from_result(result))
return game_servers
return List(
GameServer,
[
self._from_result(result)
for result in self._context.select(GameServer.get_select_by_api_key_id_string(id))
],
)
def add_game_server(self, game_server: GameServer):
self._logger.trace(__name__, f"Send SQL command: {game_server.insert_string}")

40
bot/src/bot_data/service/level_repository_service.py
View File

@ -4,8 +4,8 @@ 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.level_repository_abc import LevelRepositoryABC
from bot_data.abc.server_repository_abc import ServerRepositoryABC
from bot_data.model.level import Level
@ -41,20 +41,15 @@ class LevelRepositoryService(LevelRepositoryABC):
)
def get_levels(self) -> List[Level]:
levels = List(Level)
self._logger.trace(__name__, f"Send SQL command: {Level.get_select_all_string()}")
results = self._context.select(Level.get_select_all_string())
for result in results:
self._logger.trace(__name__, f"Get level with id {result[0]}")
levels.append(self._level_from_result(result))
return levels
return List(
Level, [self._level_from_result(result) for result in self._context.select(Level.get_select_all_string())]
)
def get_level_by_id(self, id: int) -> Level:
self._logger.trace(__name__, f"Send SQL command: {Level.get_select_by_id_string(id)}")
result = self._context.select(Level.get_select_by_id_string(id))[0]
return self._level_from_result(result)
return self._level_from_result(self._context.select(Level.get_select_by_id_string(id))[0])
def find_level_by_id(self, id: int) -> Optional[Level]:
self._logger.trace(__name__, f"Send SQL command: {Level.get_select_by_id_string(id)}")
@ -65,21 +60,19 @@ class LevelRepositoryService(LevelRepositoryABC):
return self._level_from_result(result[0])
def get_levels_by_server_id(self, server_id: int) -> List[Level]:
levels = List(Level)
self._logger.trace(
__name__,
f"Send SQL command: {Level.get_select_by_server_id_string(server_id)}",
)
results = self._context.select(Level.get_select_by_server_id_string(server_id))
for result in results:
self._logger.trace(__name__, f"Get level with id {result[0]}")
levels.append(self._level_from_result(result))
return levels
return List(
Level,
[
self._level_from_result(result)
for result in self._context.select(Level.get_select_by_server_id_string(server_id))
],
)
def find_levels_by_server_id(self, server_id: int) -> Optional[List[Level]]:
levels = List(Level)
self._logger.trace(
__name__,
f"Send SQL command: {Level.get_select_by_server_id_string(server_id)}",
@ -88,11 +81,10 @@ class LevelRepositoryService(LevelRepositoryABC):
if results is None or len(results) == 0:
return None
for result in results:
self._logger.trace(__name__, f"Get level with id {result[0]}")
levels.append(self._level_from_result(result))
return levels
return List(
Level,
[self._level_from_result(result) for result in results],
)
def add_level(self, level: Level):
self._logger.trace(__name__, f"Send SQL command: {level.insert_string}")

32
bot/src/bot_data/service/scheduled_event_repository_service.py
View File

@ -5,8 +5,8 @@ from cpl_query.extension import List
from discord import EntityType
from bot_core.logging.database_logger import DatabaseLogger
from bot_data.abc.server_repository_abc import ServerRepositoryABC
from bot_data.abc.scheduled_event_repository_abc import ScheduledEventRepositoryABC
from bot_data.abc.server_repository_abc import ServerRepositoryABC
from bot_data.model.scheduled_event import ScheduledEvent
from bot_data.model.scheduled_event_interval_enum import ScheduledEventIntervalEnum
@ -49,34 +49,34 @@ class ScheduledEventRepositoryService(ScheduledEventRepositoryABC):
)
def get_scheduled_events(self) -> List[ScheduledEvent]:
scheduled_events = List(ScheduledEvent)
self._logger.trace(__name__, f"Send SQL command: {ScheduledEvent.get_select_all_string()}")
results = self._context.select(ScheduledEvent.get_select_all_string())
for result in results:
self._logger.trace(__name__, f"Get scheduled_event with id {result[0]}")
scheduled_events.append(self._scheduled_event_from_result(result))
return scheduled_events
return List(
ScheduledEvent,
[
self._scheduled_event_from_result(result)
for result in self._context.select(ScheduledEvent.get_select_all_string())
],
)
def get_scheduled_event_by_id(self, id: int) -> ScheduledEvent:
self._logger.trace(__name__, f"Send SQL command: {ScheduledEvent.get_select_by_id_string(id)}")
result = self._context.select(ScheduledEvent.get_select_by_id_string(id))[0]
return self._scheduled_event_from_result(result)
return self._scheduled_event_from_result(self._context.select(ScheduledEvent.get_select_by_id_string(id))[0])
def get_scheduled_events_by_server_id(self, server_id: int) -> List[ScheduledEvent]:
scheduled_events = List(ScheduledEvent)
self._logger.trace(
__name__,
f"Send SQL command: {ScheduledEvent.get_select_by_server_id_string(server_id)}",
)
results = self._context.select(ScheduledEvent.get_select_by_server_id_string(server_id))
for result in results:
self._logger.trace(__name__, f"Get scheduled_event with id {result[0]}")
scheduled_events.append(self._scheduled_event_from_result(result))
return scheduled_events
return List(
ScheduledEvent,
[
self._scheduled_event_from_result(result)
for result in self._context.select(ScheduledEvent.get_select_by_server_id_string(server_id))
],
)
def add_scheduled_event(self, scheduled_event: ScheduledEvent):
self._logger.trace(__name__, f"Send SQL command: {scheduled_event.insert_string}")

36
bot/src/bot_data/service/server_config_repository_service.py
View File

@ -26,15 +26,14 @@ class ServerConfigRepositoryService(ServerConfigRepositoryABC):
self._servers = servers
def _get_team_role_ids(self, server_id: int) -> List[ServerTeamRoleIdsConfig]:
ids = List(ServerTeamRoleIdsConfig)
self._logger.trace(
__name__,
f"Send SQL command: {ServerTeamRoleIdsConfig.get_select_by_server_id_string(server_id)}",
)
results = self._context.select(ServerTeamRoleIdsConfig.get_select_by_server_id_string(server_id))
for result in results:
self._logger.trace(__name__, f"Got ServerTeamRoleIdsConfig with id {result[0]}")
ids.append(
return List(
ServerTeamRoleIdsConfig,
[
ServerTeamRoleIdsConfig(
result[1],
TeamMemberTypeEnum(result[2]),
@ -43,22 +42,23 @@ class ServerConfigRepositoryService(ServerConfigRepositoryABC):
result[5],
id=result[0],
)
)
return ids
for result in self._context.select(ServerTeamRoleIdsConfig.get_select_by_server_id_string(server_id))
],
)
def _get_afk_channel_ids(self, server_id: int) -> List[int]:
urls = List(int)
self._logger.trace(
__name__,
f"Send SQL command: {ServerAFKChannelIdsConfig.get_select_by_server_id_string(server_id)}",
)
results = self._context.select(ServerAFKChannelIdsConfig.get_select_by_server_id_string(server_id))
for result in results:
self._logger.trace(__name__, f"Got ServerAFKChannelIdsConfig with id {result[0]}")
urls.append(result[1])
return urls
return List(
int,
[
result[1]
for result in self._context.select(ServerAFKChannelIdsConfig.get_select_by_server_id_string(server_id))
],
)
def _from_result(self, result: tuple) -> ServerConfig:
return ServerConfig(
@ -102,18 +102,14 @@ class ServerConfigRepositoryService(ServerConfigRepositoryABC):
__name__,
f"Send SQL command: {ServerConfig.get_select_by_server_id_string(server_id)}",
)
result = self._context.select(ServerConfig.get_select_by_server_id_string(server_id))[0]
return self._from_result(result)
return self._from_result(self._context.select(ServerConfig.get_select_by_server_id_string(server_id))[0])
def get_server_config_by_id(self, config_id: int) -> ServerConfig:
self._logger.trace(
__name__,
f"Send SQL command: {ServerConfig.get_select_by_id_string(config_id)}",
)
result = self._context.select(ServerConfig.get_select_by_id_string(config_id))[0]
return self._from_result(result)
return self._from_result(self._context.select(ServerConfig.get_select_by_id_string(config_id))[0])
def add_server_config(self, server_config: ServerConfig):
self._logger.trace(__name__, f"Send SQL command: {server_config.insert_string}")

15
bot/src/bot_data/service/server_repository_service.py
View File

@ -25,12 +25,15 @@ class ServerRepositoryService(ServerRepositoryABC):
ServerRepositoryABC.__init__(self)
def get_servers(self) -> List[Server]:
servers = List(Server)
self._logger.trace(__name__, f"Send SQL command: {Server.get_select_all_string()}")
results = self._context.select(Server.get_select_all_string())
for result in results:
servers.append(Server(result[1], result[2], result[3], id=result[0]))
servers = List(
Server,
[
Server(result[1], result[2], result[3], id=result[0])
for result in self._context.select(Server.get_select_all_string())
],
)
self._cache.add_servers(servers)
return servers
@ -96,9 +99,7 @@ class ServerRepositoryService(ServerRepositoryABC):
if result is None or len(result) == 0:
return None
result = result[0]
return Server(result[1], result[2], result[3], id=result[0])
return Server(result[0][1], result[0][2], result[0][3], id=result[0][0])
def add_server(self, server: Server):
self._logger.trace(__name__, f"Send SQL command: {server.insert_string}")

42
bot/src/bot_data/service/short_role_name_repository_service.py
View File

@ -43,14 +43,15 @@ class ShortRoleNameRepositoryService(ShortRoleNameRepositoryABC):
)
def get_short_role_names(self) -> List[ShortRoleName]:
short_role_names = List(ShortRoleName)
self._logger.trace(__name__, f"Send SQL command: {ShortRoleName.get_select_all_string()}")
results = self._context.select(ShortRoleName.get_select_all_string())
for result in results:
self._logger.trace(__name__, f"Get short_role_name with id {result[0]}")
short_role_names.append(self._short_role_name_from_result(result))
return short_role_names
return List(
ShortRoleName,
[
self._short_role_name_from_result(result)
for result in self._context.select(ShortRoleName.get_select_all_string())
],
)
def get_short_role_name_by_id(self, id: int) -> ShortRoleName:
self._logger.trace(__name__, f"Send SQL command: {ShortRoleName.get_select_by_id_string(id)}")
@ -59,31 +60,30 @@ class ShortRoleNameRepositoryService(ShortRoleNameRepositoryABC):
return self._short_role_name_from_result(result)
def find_short_role_names_by_role_id(self, role_id: int) -> List[ShortRoleName]:
short_role_names = List(ShortRoleName)
self._logger.trace(
__name__,
f"Send SQL command: {ShortRoleName.get_select_by_role_id_string(role_id)}",
)
results = self._context.select(ShortRoleName.get_select_by_role_id_string(role_id))
for result in results:
self._logger.trace(__name__, f"Get short_role_name with id {result[0]}")
short_role_names.append(self._short_role_name_from_result(result))
return short_role_names
return List(
ShortRoleName,
[
self._short_role_name_from_result(result)
for result in self._context.select(ShortRoleName.get_select_by_role_id_string(role_id))
],
)
def get_short_role_names_by_server_id(self, server_id: int) -> List[ShortRoleName]:
short_role_names = List(ShortRoleName)
self._logger.trace(
__name__,
f"Send SQL command: {ShortRoleName.get_select_by_server_id_string(server_id)}",
)
results = self._context.select(ShortRoleName.get_select_by_server_id_string(server_id))
for result in results:
self._logger.trace(__name__, f"Get short_role_name with id {result[0]}")
short_role_names.append(self._short_role_name_from_result(result))
return short_role_names
return List(
ShortRoleName,
[
self._short_role_name_from_result(result)
for result in self._context.select(ShortRoleName.get_select_by_server_id_string(server_id))
],
)
def add_short_role_name(self, short_role_name: ShortRoleName):
self._logger.trace(__name__, f"Send SQL command: {short_role_name.insert_string}")

19
bot/src/bot_data/service/steam_special_offer_repository_service.py
View File

@ -42,23 +42,24 @@ class SteamSpecialOfferRepositoryService(SteamSpecialOfferRepositoryABC):
)
def get_steam_special_offers(self) -> List[SteamSpecialOffer]:
steam_special_offers = List(SteamSpecialOffer)
self._logger.trace(__name__, f"Send SQL command: {SteamSpecialOffer.get_select_all_string()}")
results = self._context.select(SteamSpecialOffer.get_select_all_string())
for result in results:
self._logger.trace(__name__, f"Get steam_special_offer with id {result[0]}")
steam_special_offers.append(self._steam_special_offer_from_result(result))
return steam_special_offers
return List(
SteamSpecialOffer,
[
self._steam_special_offer_from_result(result)
for result in self._context.select(SteamSpecialOffer.get_select_all_string())
],
)
def get_steam_special_offer_by_name(self, name: str) -> SteamSpecialOffer:
self._logger.trace(
__name__,
f"Send SQL command: {SteamSpecialOffer.get_select_by_name_string(name)}",
)
result = self._context.select(SteamSpecialOffer.get_select_by_name_string(name))[0]
return self._steam_special_offer_from_result(result)
return self._steam_special_offer_from_result(
self._context.select(SteamSpecialOffer.get_select_by_name_string(name))[0]
)
def add_steam_special_offer(self, steam_special_offer: SteamSpecialOffer):
self._logger.trace(__name__, f"Send SQL command: {steam_special_offer.insert_string}")

22
bot/src/bot_data/service/technician_config_repository_service.py
View File

@ -18,27 +18,17 @@ class TechnicianConfigRepositoryService(TechnicianConfigRepositoryABC):
self._context = db_context
def _get_technician_ids(self) -> List[int]:
ids = List(int)
self._logger.trace(__name__, f"Send SQL command: {TechnicianIdConfig.get_select_all_string()}")
results = self._context.select(TechnicianIdConfig.get_select_all_string())
for result in results:
self._logger.trace(__name__, f"Got TechnicianId with id {result[0]}")
ids.append(result[1])
return ids
return List(int, [config[1] for config in self._context.select(TechnicianIdConfig.get_select_all_string())])
def _get_technician_ping_urls(self) -> List[str]:
urls = List(str)
self._logger.trace(
__name__,
f"Send SQL command: {TechnicianPingUrlConfig.get_select_all_string()}",
)
results = self._context.select(TechnicianPingUrlConfig.get_select_all_string())
for result in results:
self._logger.trace(__name__, f"Got TechnicianPingUrl with id {result[0]}")
urls.append(result[1])
return urls
return List(
str, [ping_url[1] for ping_url in self._context.select(TechnicianPingUrlConfig.get_select_all_string())]
)
def _from_result(self, result: tuple) -> TechnicianConfig:
return TechnicianConfig(
@ -64,9 +54,7 @@ class TechnicianConfigRepositoryService(TechnicianConfigRepositoryABC):
def get_technician_config(self) -> TechnicianConfig:
self._logger.trace(__name__, f"Send SQL command: {TechnicianConfig.get_select_all_string()}")
result = self._context.select(TechnicianConfig.get_select_all_string())[0]
return self._from_result(result)
return self._from_result(self._context.select(TechnicianConfig.get_select_all_string())[0])
def add_technician_config(self, technician_config: TechnicianConfig):
self._logger.trace(__name__, f"Send SQL command: {technician_config.insert_string}")

51
bot/src/bot_data/service/user_game_ident_repository_service.py
View File

@ -39,46 +39,47 @@ class UserGameIdentRepositoryService(UserGameIdentRepositoryABC):
)
def get_user_game_idents(self) -> List[UserGameIdent]:
joins = List(UserGameIdent)
self._logger.trace(
__name__,
f"Send SQL command: {UserGameIdent.get_select_all_string()}",
)
results = self._context.select(UserGameIdent.get_select_all_string())
for result in results:
self._logger.trace(__name__, f"Get UserGameIdent with id {result[0]}")
joins.append(self._from_result(result))
return joins
return List(
UserGameIdent,
[
self._from_result(game_ident)
for game_ident in self._context.select(UserGameIdent.get_select_all_string())
],
)
def get_user_game_idents_by_game_server_id(self, game_server_id: int) -> List[UserGameIdent]:
joins = List(UserGameIdent)
self._logger.trace(
__name__,
f"Send SQL command: {UserGameIdent.get_select_by_game_server_id_string(game_server_id)}",
)
results = self._context.select(UserGameIdent.get_select_by_game_server_id_string(game_server_id))
for result in results:
self._logger.trace(__name__, f"Get UserGameIdent with id {result[0]}")
joins.append(self._from_result(result))
return joins
return List(
UserGameIdent,
[
self._from_result(game_ident)
for game_ident in self._context.select(
UserGameIdent.get_select_by_game_server_id_string(game_server_id)
)
],
)
def get_user_game_ident_by_id(self, id: int) -> UserGameIdent:
self._logger.trace(
__name__,
f"Send SQL command: {UserGameIdent.get_select_by_id_string(id)}",
)
result = self._context.select(UserGameIdent.get_select_by_id_string(id))[0]
return self._from_result(result)
return self._from_result(self._context.select(UserGameIdent.get_select_by_id_string(id))[0])
def get_user_game_ident_by_ident(self, ident: str) -> UserGameIdent:
self._logger.trace(
__name__,
f"Send SQL command: {UserGameIdent.get_select_by_ident_string(ident)}",
)
result = self._context.select(UserGameIdent.get_select_by_ident_string(ident))[0]
return self._from_result(result)
return self._from_result(self._context.select(UserGameIdent.get_select_by_ident_string(ident))[0])
def find_user_game_ident_by_ident(self, ident: str) -> Optional[UserGameIdent]:
self._logger.trace(
@ -89,20 +90,20 @@ class UserGameIdentRepositoryService(UserGameIdentRepositoryABC):
if len(result) == 0:
return None
result = result[0]
return self._from_result(result)
return self._from_result(result[0])
def get_user_game_idents_by_user_id(self, user_id: int) -> List[UserGameIdent]:
joins = List(UserGameIdent)
self._logger.trace(
__name__,
f"Send SQL command: {UserGameIdent.get_select_by_user_id_string(user_id)}",
)
results = self._context.select(UserGameIdent.get_select_by_user_id_string(user_id))
for result in results:
joins.append(self._from_result(result))
return joins
return List(
UserGameIdent,
[
self._from_result(game_ident)
for game_ident in self._context.select(UserGameIdent.get_select_by_user_id_string(user_id))
],
)
def add_user_game_ident(self, user_game_ident: UserGameIdent):
self._logger.trace(__name__, f"Send SQL command: {user_game_ident.insert_string}")

47
bot/src/bot_data/service/user_joined_game_server_repository_service.py
View File

@ -39,45 +39,45 @@ class UserJoinedGameServerRepositoryService(UserJoinedGameServerRepositoryABC):
)
def get_user_joined_game_servers(self) -> List[UserJoinedGameServer]:
joins = List(UserJoinedGameServer)
self._logger.trace(
__name__,
f"Send SQL command: {UserJoinedGameServer.get_select_all_string()}",
)
results = self._context.select(UserJoinedGameServer.get_select_all_string())
for result in results:
self._logger.trace(__name__, f"Get user-joined-game-server with id {result[0]}")
joins.append(self._from_result(result))
return joins
return List(
UserJoinedGameServer,
[self._from_result(join) for join in self._context.select(UserJoinedGameServer.get_select_all_string())],
)
def get_user_joined_game_server_by_id(self, id: int) -> UserJoinedGameServer:
self._logger.trace(
__name__,
f"Send SQL command: {UserJoinedGameServer.get_select_by_id_string(id)}",
)
result = self._context.select(UserJoinedGameServer.get_select_by_id_string(id))[0]
return self._from_result(result)
return self._from_result(self._context.select(UserJoinedGameServer.get_select_by_id_string(id))[0])
def get_user_joined_game_servers_by_user_id(self, user_id: int) -> List[UserJoinedGameServer]:
joins = List(UserJoinedGameServer)
self._logger.trace(
__name__,
f"Send SQL command: {UserJoinedGameServer.get_select_by_user_id_string(user_id)}",
)
results = self._context.select(UserJoinedGameServer.get_select_by_user_id_string(user_id))
for result in results:
joins.append(self._from_result(result))
return joins
return List(
UserJoinedGameServer,
[
self._from_result(join)
for join in self._context.select(UserJoinedGameServer.get_select_by_user_id_string(user_id))
],
)
def get_active_user_joined_game_server_by_user_id(self, user_id: int) -> UserJoinedGameServer:
self._logger.trace(
__name__,
f"Send SQL command: {UserJoinedGameServer.get_select_by_user_id_string(user_id)}",
)
result = self._context.select(UserJoinedGameServer.get_select_active_by_user_id_string(user_id))[0]
return self._from_result(result)
return self._from_result(
self._context.select(UserJoinedGameServer.get_select_active_by_user_id_string(user_id))[0]
)
def find_active_user_joined_game_server_by_user_id(self, user_id: int) -> Optional[UserJoinedGameServer]:
self._logger.trace(
@ -88,22 +88,21 @@ class UserJoinedGameServerRepositoryService(UserJoinedGameServerRepositoryABC):
if result is None or len(result) == 0:
return None
result = result[0]
return self._from_result(result)
return self._from_result(result[0])
def find_active_user_joined_game_servers_by_user_id(self, user_id: int) -> List[Optional[UserJoinedGameServer]]:
self._logger.trace(
__name__,
f"Send SQL command: {UserJoinedGameServer.get_select_active_by_user_id_string(user_id)}",
)
result = List(UserJoinedGameServer)
db_results = self._context.select(UserJoinedGameServer.get_select_active_by_user_id_string(user_id))
for db_result in db_results:
result.append(self._from_result(db_result))
return result
return List(
UserJoinedGameServer,
[
self._from_result(join)
for join in self._context.select(UserJoinedGameServer.get_select_active_by_user_id_string(user_id))
],
)
def add_user_joined_game_server(self, user_joined_game_server: UserJoinedGameServer):
self._logger.trace(__name__, f"Send SQL command: {user_joined_game_server.insert_string}")

101
bot/src/bot_data/service/user_joined_server_repository_service.py
View File

@ -23,89 +23,51 @@ class UserJoinedServerRepositoryService(UserJoinedServerRepositoryABC):
UserJoinedServerRepositoryABC.__init__(self)
def get_user_joined_servers(self) -> List[UserJoinedServer]:
joins = List(UserJoinedServer)
self._logger.trace(__name__, f"Send SQL command: {UserJoinedServer.get_select_all_string()}")
results = self._context.select(UserJoinedServer.get_select_all_string())
for result in results:
self._logger.trace(__name__, f"Get user-joined-server with id {result[0]}")
joins.append(
UserJoinedServer(
self._users.get_user_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
)
def __from_result(self, result: tuple) -> UserJoinedServer:
return UserJoinedServer(
self._users.get_user_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
return joins
def get_user_joined_servers(self) -> List[UserJoinedServer]:
self._logger.trace(__name__, f"Send SQL command: {UserJoinedServer.get_select_all_string()}")
return List(
UserJoinedServer,
[self.__from_result(join) for join in self._context.select(UserJoinedServer.get_select_all_string())],
)
def get_user_joined_server_by_id(self, id: int) -> UserJoinedServer:
self._logger.trace(
__name__,
f"Send SQL command: {UserJoinedServer.get_select_by_id_string(id)}",
)
result = self._context.select(UserJoinedServer.get_select_by_id_string(id))[0]
return UserJoinedServer(
self._users.get_user_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
def get_user_joined_server_by_server_id(self, server_id: int) -> UserJoinedServer:
self._logger.trace(
__name__,
f"Send SQL command: {UserJoinedServer.get(id)}",
)
result = self._context.select(UserJoinedServer.get_select_by_id_string(id))[0]
return UserJoinedServer(
self._users.get_user_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
return self.__from_result(self._context.select(UserJoinedServer.get_select_by_id_string(id))[0])
def get_user_joined_servers_by_user_id(self, user_id: int) -> List[UserJoinedServer]:
joins = List(UserJoinedServer)
self._logger.trace(
__name__,
f"Send SQL command: {UserJoinedServer.get_select_by_user_id_string(user_id)}",
)
results = self._context.select(UserJoinedServer.get_select_by_user_id_string(user_id))
for result in results:
joins.append(
UserJoinedServer(
self._users.get_user_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
)
return joins
return List(
UserJoinedServer,
[
self.__from_result(join)
for join in self._context.select(UserJoinedServer.get_select_by_user_id_string(user_id))
],
)
def get_active_user_joined_server_by_user_id(self, user_id: int) -> UserJoinedServer:
self._logger.trace(
__name__,
f"Send SQL command: {UserJoinedServer.get_select_by_user_id_string(user_id)}",
)
result = self._context.select(UserJoinedServer.get_select_active_by_user_id_string(user_id))[0]
return UserJoinedServer(
self._users.get_user_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
return self.__from_result(
self._context.select(UserJoinedServer.get_select_active_by_user_id_string(user_id))[0]
)
def find_active_user_joined_server_by_user_id(self, user_id: int) -> Optional[UserJoinedServer]:
@ -117,16 +79,7 @@ class UserJoinedServerRepositoryService(UserJoinedServerRepositoryABC):
if result is None or len(result) == 0:
return None
result = result[0]
return UserJoinedServer(
self._users.get_user_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
return self.__from_result(result[0])
def add_user_joined_server(self, user_joined_server: UserJoinedServer):
self._logger.trace(__name__, f"Send SQL command: {user_joined_server.insert_string}")

103
bot/src/bot_data/service/user_joined_voice_channel_repository_service.py
View File

@ -25,77 +25,55 @@ class UserJoinedVoiceChannelRepositoryService(UserJoinedVoiceChannelRepositoryAB
UserJoinedVoiceChannelRepositoryABC.__init__(self)
def __from_result(self, result: tuple) -> UserJoinedVoiceChannel:
return UserJoinedVoiceChannel(
self._users.get_user_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
def get_user_joined_voice_channels(self) -> List[UserJoinedVoiceChannel]:
joins = List(UserJoinedVoiceChannel)
self._logger.trace(
__name__,
f"Send SQL command: {UserJoinedVoiceChannel.get_select_all_string()}",
)
results = self._context.select(UserJoinedVoiceChannel.get_select_all_string())
for result in results:
self._logger.trace(__name__, f"Get user-joined-voice-channel with id {result[0]}")
joins.append(
UserJoinedVoiceChannel(
self._users.get_user_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
)
return joins
return List(
UserJoinedVoiceChannel,
[self.__from_result(join) for join in self._context.select(UserJoinedVoiceChannel.get_select_all_string())],
)
def get_user_joined_voice_channel_by_id(self, id: int) -> UserJoinedVoiceChannel:
self._logger.trace(
__name__,
f"Send SQL command: {UserJoinedVoiceChannel.get_select_by_id_string(id)}",
)
result = self._context.select(UserJoinedVoiceChannel.get_select_by_id_string(id))[0]
return UserJoinedVoiceChannel(
self._users.get_user_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
return self.__from_result(self._context.select(UserJoinedVoiceChannel.get_select_by_id_string(id))[0])
def get_user_joined_voice_channels_by_user_id(self, user_id: int) -> List[UserJoinedVoiceChannel]:
joins = List(UserJoinedVoiceChannel)
self._logger.trace(
__name__,
f"Send SQL command: {UserJoinedVoiceChannel.get_select_by_user_id_string(user_id)}",
)
results = self._context.select(UserJoinedVoiceChannel.get_select_by_user_id_string(user_id))
for result in results:
joins.append(
UserJoinedVoiceChannel(
self._users.get_user_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
)
return joins
return List(
UserJoinedVoiceChannel,
[
self.__from_result(join)
for join in self._context.select(UserJoinedVoiceChannel.get_select_by_user_id_string(user_id))
],
)
def get_active_user_joined_voice_channel_by_user_id(self, user_id: int) -> UserJoinedVoiceChannel:
self._logger.trace(
__name__,
f"Send SQL command: {UserJoinedVoiceChannel.get_select_by_user_id_string(user_id)}",
)
result = self._context.select(UserJoinedVoiceChannel.get_select_active_by_user_id_string(user_id))[0]
return UserJoinedVoiceChannel(
self._users.get_user_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
return self.__from_result(
self._context.select(UserJoinedVoiceChannel.get_select_active_by_user_id_string(user_id))[0]
)
def find_active_user_joined_voice_channel_by_user_id(self, user_id: int) -> Optional[UserJoinedVoiceChannel]:
@ -107,38 +85,21 @@ class UserJoinedVoiceChannelRepositoryService(UserJoinedVoiceChannelRepositoryAB
if result is None or len(result) == 0:
return None
result = result[0]
return UserJoinedVoiceChannel(
self._users.get_user_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
return self.__from_result(result[0])
def find_active_user_joined_voice_channels_by_user_id(self, user_id: int) -> List[Optional[UserJoinedVoiceChannel]]:
self._logger.trace(
__name__,
f"Send SQL command: {UserJoinedVoiceChannel.get_select_active_by_user_id_string(user_id)}",
)
result = List(UserJoinedVoiceChannel)
db_results = self._context.select(UserJoinedVoiceChannel.get_select_active_by_user_id_string(user_id))
for db_result in db_results:
result.append(
UserJoinedVoiceChannel(
self._users.get_user_by_id(db_result[1]),
db_result[2],
db_result[3],
db_result[4],
db_result[5],
id=db_result[0],
)
)
return result
return List(
UserJoinedVoiceChannel,
[
self.__from_result(join)
for join in self._context.select(UserJoinedVoiceChannel.get_select_active_by_user_id_string(user_id))
],
)
def add_user_joined_voice_channel(self, user_joined_voice_channel: UserJoinedVoiceChannel):
self._logger.trace(__name__, f"Send SQL command: {user_joined_voice_channel.insert_string}")

27
bot/src/bot_data/service/user_message_count_per_hour_repository_service.py
View File

@ -44,39 +44,30 @@ class UserMessageCountPerHourRepositoryService(UserMessageCountPerHourRepository
)
def get_user_message_count_per_hours(self) -> List[UserMessageCountPerHour]:
umcphs = List(UserMessageCountPerHour)
self._logger.trace(
__name__,
f"Send SQL command: {UserMessageCountPerHour.get_select_all_string()}",
)
results = self._context.select(UserMessageCountPerHour.get_select_all_string())
for result in results:
self._logger.trace(__name__, f"Get user message count per hour with id {result[0]}")
umcphs.append(self._from_result(result))
return umcphs
return List(
UserMessageCountPerHour,
[
self._from_result(umcphs)
for umcphs in self._context.select(UserMessageCountPerHour.get_select_all_string())
],
)
def find_user_message_count_per_hour_by_user_id(self, user_id: int) -> List[Optional[UserMessageCountPerHour]]:
umcphs = List(UserMessageCountPerHour)
sql = UserMessageCountPerHour.get_select_by_user_id_string(user_id)
self._logger.trace(__name__, f"Send SQL command: {sql}")
results = self._context.select(sql)
if results is None or len(results) == 0:
return umcphs
for result in results:
self._logger.trace(__name__, f"Get user message count per hour with id {result[0]}")
umcphs.append(self._from_result(result))
return umcphs
return List(UserMessageCountPerHour, [self._from_result(umcphs) for umcphs in self._context.select(sql)])
def get_user_message_count_per_hour_by_user_id_and_date(
self, user_id: int, date: datetime
) -> UserMessageCountPerHour:
sql = UserMessageCountPerHour.get_select_by_user_id_and_date_string(user_id, date)
self._logger.trace(__name__, f"Send SQL command: {sql}")
result = self._context.select(sql)[0]
return self._from_result(result)
return self._from_result(self._context.select(sql)[0])
def find_user_message_count_per_hour_by_user_id_and_date(
self, user_id: int, date: datetime

45
bot/src/bot_data/service/user_repository_service.py
View File

@ -1,4 +1,3 @@
import datetime
from typing import Optional
from cpl_core.database.context import DatabaseContextABC
@ -38,14 +37,8 @@ class UserRepositoryService(UserRepositoryABC):
)
def get_users(self) -> List[User]:
users = List(User)
self._logger.trace(__name__, f"Send SQL command: {User.get_select_all_string()}")
results = self._context.select(User.get_select_all_string())
for result in results:
self._logger.trace(__name__, f"Get user with id {result[0]}")
users.append(self._from_result(result))
return users
return List(User, [self._from_result(user) for user in self._context.select(User.get_select_all_string())])
def get_user_by_id(self, id: int) -> User:
self._logger.trace(__name__, f"Send SQL command: {User.get_select_by_id_string(id)}")
@ -59,42 +52,40 @@ class UserRepositoryService(UserRepositoryABC):
if result is None or len(result) == 0:
return None
result = result[0]
return self._from_result(result)
return self._from_result(result[0])
def get_users_by_discord_id(self, discord_id: int) -> List[User]:
users = List(User)
self._logger.trace(
__name__,
f"Send SQL command: {User.get_select_by_discord_id_string(discord_id)}",
)
results = self._context.select(User.get_select_by_discord_id_string(discord_id))
for result in results:
users.append(self._from_result(result))
return users
return List(
User,
[
self._from_result(user)
for user in self._context.select(User.get_select_by_discord_id_string(discord_id))
],
)
def get_users_by_server_id(self, server_id: int) -> List[User]:
users = List(User)
self._logger.trace(
__name__,
f"Send SQL command: {User.get_select_by_server_id_string(server_id)}",
)
results = self._context.select(User.get_select_by_server_id_string(server_id))
for result in results:
users.append(self._from_result(result))
return users
return List(
User,
[self._from_result(user) for user in self._context.select(User.get_select_by_server_id_string(server_id))],
)
def get_user_by_discord_id_and_server_id(self, discord_id: int, server_id: int) -> User:
self._logger.trace(
__name__,
f"Send SQL command: {User.get_select_by_discord_id_and_server_id_string(discord_id, server_id)}",
)
result = self._context.select(User.get_select_by_discord_id_and_server_id_string(discord_id, server_id))[0]
return self._from_result(result)
return self._from_result(
self._context.select(User.get_select_by_discord_id_and_server_id_string(discord_id, server_id))[0]
)
def find_user_by_discord_id_and_server_id(self, discord_id: int, server_id: int) -> Optional[User]:
self._logger.trace(
@ -105,9 +96,7 @@ class UserRepositoryService(UserRepositoryABC):
if result is None or len(result) == 0:
return None
result = result[0]
return self._from_result(result)
return self._from_result(result[0])
def add_user(self, user: User):
self._logger.trace(__name__, f"Send SQL command: {user.insert_string}")

25
bot/src/bot_data/service/user_warnings_repository_service.py
View File

@ -46,30 +46,29 @@ class UserWarningsRepositoryService(UserWarningsRepositoryABC):
)
def get_user_warnings(self) -> List[UserWarnings]:
warnings = List(UserWarnings)
self._logger.trace(__name__, f"Send SQL command: {UserWarnings.get_select_all_string()}")
results = self._context.select(UserWarnings.get_select_all_string())
for result in results:
warnings.append(self._from_result(result))
return warnings
return List(
UserWarnings,
[self._from_result(warning) for warning in self._context.select(UserWarnings.get_select_all_string())],
)
def get_user_warnings_by_id(self, id: int) -> UserWarnings:
self._logger.trace(__name__, f"Send SQL command: {UserWarnings.get_select_by_id_string(id)}")
result = self._context.select(UserWarnings.get_select_by_id_string(id))[0]
return self._from_result(result)
return self._from_result(self._context.select(UserWarnings.get_select_by_id_string(id))[0])
def get_user_warnings_by_user_id(self, user_id: int) -> List[UserWarnings]:
warnings = List(UserWarnings)
self._logger.trace(
__name__,
f"Send SQL command: {UserWarnings.get_select_by_user_id_string(user_id)}",
)
results = self._context.select(UserWarnings.get_select_by_user_id_string(user_id))
for result in results:
warnings.append(self._from_result(result))
return warnings
return List(
UserWarnings,
[
self._from_result(warning)
for warning in self._context.select(UserWarnings.get_select_by_user_id_string(user_id))
],
)
def add_user_warnings(self, user_warnings: UserWarnings):
self._logger.trace(__name__, f"Send SQL command: {user_warnings.insert_string}")

4
bot/src/bot_graphql/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_graphql"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

4
bot/src/bot_graphql/abc/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_graphql.abc"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports:
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

2
bot/src/bot_graphql/abc/query_abc.py
View File

@ -12,6 +12,7 @@ from bot_api.exception.service_exception import ServiceException
from bot_api.route.route import Route
from bot_core.configuration.feature_flags_enum import FeatureFlagsEnum
from bot_core.environment_variables import MAINTENANCE
from bot_core.service.permission_service import PermissionService
from bot_data.model.achievement import Achievement
from bot_data.model.auth_role_enum import AuthRoleEnum
from bot_data.model.auth_user import AuthUser
@ -32,7 +33,6 @@ from bot_data.model.user_role_enum import UserRoleEnum
from bot_graphql.abc.filter_abc import FilterABC
from bot_graphql.filter.page import Page
from bot_graphql.filter.sort import Sort
from modules.permission.service.permission_service import PermissionService
class QueryABC(ObjectType):

2
bot/src/bot_graphql/bot-graphql.json
View File

@ -4,7 +4,7 @@
"Version": {
"Major": "1",
"Minor": "2",
"Micro": "2"
"Micro": "3"
},
"Author": "Sven Heidemann",
"AuthorEmail": "sven.heidemann@sh-edraft.de",

4
bot/src/bot_graphql/filter/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_graphql.filter"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports:
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

1
bot/src/bot_graphql/graphql/query.gql
View File

@ -52,4 +52,5 @@ type Query {
discord: Discord
hasFeatureFlag(flag: String): FeatureFlag
featureFlags: [FeatureFlag]
}

1
bot/src/bot_graphql/graphql/server.gql
View File

@ -40,6 +40,7 @@ type Server implements TableWithHistoryQuery {
config: ServerConfig
hasFeatureFlag(flag: String): FeatureFlag
featureFlags: [FeatureFlag]
statistic(date: String): ServerStatistic

1
bot/src/bot_graphql/graphql/user.gql
View File

@ -7,6 +7,7 @@ type User implements TableWithHistoryQuery {
reactionCount: Int
birthday: String
ontime: Float
gameOntime: Float
level: Level
joinedServers(filter: UserJoinedServerFilter, page: Page, sort: Sort): [UserJoinedServer]

4
bot/src/bot_graphql/model/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_graphql.model"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

4
bot/src/bot_graphql/mutations/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_graphql.mutations"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports:
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

2
bot/src/bot_graphql/mutations/achievement_mutation.py
View File

@ -6,7 +6,7 @@ from bot_data.abc.server_repository_abc import ServerRepositoryABC
from bot_data.model.achievement import Achievement
from bot_data.model.user_role_enum import UserRoleEnum
from bot_graphql.abc.query_abc import QueryABC
from modules.permission.service.permission_service import PermissionService
from bot_core.service.permission_service import PermissionService
class AchievementMutation(QueryABC):

2
bot/src/bot_graphql/mutations/scheduled_event_mutation.py
View File

@ -10,7 +10,7 @@ from bot_data.model.scheduled_event import ScheduledEvent
from bot_data.model.scheduled_event_interval_enum import ScheduledEventIntervalEnum
from bot_data.model.user_role_enum import UserRoleEnum
from bot_graphql.abc.query_abc import QueryABC
from modules.permission.service.permission_service import PermissionService
from bot_core.service.permission_service import PermissionService
class ScheduledEventMutation(QueryABC):

9
bot/src/bot_graphql/mutations/server_config_mutation.py
View File

@ -13,6 +13,7 @@ from bot_data.model.server_team_role_ids_config import ServerTeamRoleIdsConfig
from bot_data.model.team_member_type_enum import TeamMemberTypeEnum
from bot_data.model.user_role_enum import UserRoleEnum
from bot_graphql.abc.query_abc import QueryABC
from bot_core.service.permission_service import PermissionService
class ServerConfigMutation(QueryABC):
@ -24,6 +25,7 @@ class ServerConfigMutation(QueryABC):
server_configs: ServerConfigRepositoryABC,
db: DatabaseContextABC,
config_service: ConfigService,
permissions: PermissionService,
):
QueryABC.__init__(self, "ServerConfigMutation")
@ -33,6 +35,7 @@ class ServerConfigMutation(QueryABC):
self._server_configs = server_configs
self._db = db
self._config_service = config_service
self._permissions = permissions
self.set_field("updateServerConfig", self.resolve_update_server_config)
@ -175,11 +178,13 @@ class ServerConfigMutation(QueryABC):
def _update_team_role_ids(self, new_config: ServerConfig):
old_config = self._server_configs.get_server_config_by_server(new_config.server.id)
has_update = False
for role_id in old_config.team_role_ids:
if role_id.role_id in new_config.team_role_ids.select(lambda x: int(x.role_id)):
continue
self._server_configs.delete_server_team_role_id_config(role_id)
has_update = True
for role_id in new_config.team_role_ids:
guild = self._bot.get_guild(new_config.server.discord_id)
@ -192,3 +197,7 @@ class ServerConfigMutation(QueryABC):
continue
self._server_configs.add_server_team_role_id_config(role_id)
has_update = True
if has_update:
self._permissions.reset_cache()

2
bot/src/bot_graphql/mutations/short_role_name_mutation.py
View File

@ -6,7 +6,7 @@ from bot_data.abc.short_role_name_repository_abc import ShortRoleNameRepositoryA
from bot_data.model.short_role_name import ShortRoleName
from bot_data.model.user_role_enum import UserRoleEnum
from bot_graphql.abc.query_abc import QueryABC
from modules.permission.service.permission_service import PermissionService
from bot_core.service.permission_service import PermissionService
class ShortRoleNameMutation(QueryABC):

9
bot/src/bot_graphql/mutations/technician_config_mutation.py
View File

@ -16,6 +16,7 @@ from bot_data.model.technician_ping_url_config import TechnicianPingUrlConfig
from bot_data.model.user_role_enum import UserRoleEnum
from bot_data.service.technician_config_seeder import TechnicianConfigSeeder
from bot_graphql.abc.query_abc import QueryABC
from bot_core.service.permission_service import PermissionService
class TechnicianConfigMutation(QueryABC):
@ -30,6 +31,7 @@ class TechnicianConfigMutation(QueryABC):
config_service: ConfigService,
tech_seeder: TechnicianConfigSeeder,
client_utils: ClientUtilsABC,
permissions: PermissionService,
):
QueryABC.__init__(self, "TechnicianConfigMutation")
@ -42,6 +44,7 @@ class TechnicianConfigMutation(QueryABC):
self._config_service = config_service
self._tech_seeder = tech_seeder
self._client_utils = client_utils
self._permissions = permissions
self.set_field("updateTechnicianConfig", self.resolve_update_technician_config)
@ -126,11 +129,13 @@ class TechnicianConfigMutation(QueryABC):
def _update_technician_ids(self, new_config: TechnicianConfig):
old_config = self._technician_configs.get_technician_config()
has_update = False
for technician_id in old_config.technician_ids:
if technician_id in new_config.technician_ids:
continue
self._technician_configs.delete_technician_id_config(TechnicianIdConfig(technician_id))
has_update = True
for technician_id in new_config.technician_ids:
user = self._bot.get_user(technician_id)
@ -142,3 +147,7 @@ class TechnicianConfigMutation(QueryABC):
continue
self._technician_configs.add_technician_id_config(TechnicianIdConfig(technician_id))
has_update = True
if has_update:
self._permissions.reset_cache()

2
bot/src/bot_graphql/mutations/user_joined_game_server_mutation.py
View File

@ -24,7 +24,7 @@ from bot_data.model.server_config import ServerConfig
from bot_data.model.user_joined_game_server import UserJoinedGameServer
from bot_data.model.user_role_enum import UserRoleEnum
from bot_graphql.abc.query_abc import QueryABC
from modules.permission.service.permission_service import PermissionService
from bot_core.service.permission_service import PermissionService
class UserJoinedGameServerMutation(QueryABC):

2
bot/src/bot_graphql/mutations/user_mutation.py
View File

@ -13,7 +13,7 @@ from bot_data.model.user_role_enum import UserRoleEnum
from bot_graphql.abc.query_abc import QueryABC
from modules.base.service.user_warnings_service import UserWarningsService
from modules.level.service.level_service import LevelService
from modules.permission.service.permission_service import PermissionService
from bot_core.service.permission_service import PermissionService
class UserMutation(QueryABC):

4
bot/src/bot_graphql/queries/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_graphql.queries"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports:
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

4
bot/src/bot_graphql/queries/discord/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "bot_graphql.queries.discord"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports:
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

14
bot/src/bot_graphql/queries/server_query.py
View File

@ -116,6 +116,10 @@ class ServerQuery(DataQueryWithHistoryABC):
"hasFeatureFlag",
lambda server, *_, **kwargs: self._resolve_has_feature_flag(server, *_, **kwargs),
)
self.set_field(
"featureFlags",
lambda server, *_, **kwargs: self._resolve_feature_flags(server, *_),
)
self.set_field("statistic", lambda server, *_, **kwargs: ServerStatistics(server, kwargs))
@staticmethod
@ -142,3 +146,13 @@ class ServerQuery(DataQueryWithHistoryABC):
"key": kwargs["flag"],
"value": FeatureFlagsSettings.get_flag_from_dict(settings.feature_flags, FeatureFlagsEnum(kwargs["flag"])),
}
def _resolve_feature_flags(self, server: Server, *_) -> list[dict]:
settings: ServerConfig = self._config.get_configuration(f"ServerConfig_{server.discord_id}")
return [
{
"key": flag,
"value": FeatureFlagsSettings.get_flag_from_dict(settings.feature_flags, FeatureFlagsEnum(flag)),
}
for flag in [e.value for e in FeatureFlagsEnum]
]

2
bot/src/bot_graphql/queries/server_statistic_query.py
View File

@ -132,7 +132,7 @@ class ServerStatisticQuery(QueryABC):
def _resolve_game_server_ontime(self, server, *_):
query = f"""
SELECT ROUND(SUM(TIME_TO_SEC(TIMEDIFF(UserJoinedGameServer.LeavedOn, UserJoinedGameServer.JoinedOn)) / 3600),{server.server.id}) FROM UserJoinedGameServer
SELECT ROUND(SUM(TIME_TO_SEC(TIMEDIFF(UserJoinedGameServer.LeavedOn, UserJoinedGameServer.JoinedOn)) / 3600),2) FROM UserJoinedGameServer
INNER JOIN Users ON UserJoinedGameServer.UserId = Users.UserId
WHERE Users.ServerId = {server.server.id}
AND UserJoinedGameServer.CreatedAt >= "{self._get_date(**server.kwargs)}";

52
bot/src/bot_graphql/queries/user_query.py
View File

@ -2,6 +2,7 @@ from cpl_core.database.context import DatabaseContextABC
from cpl_discord.service import DiscordBotServiceABC
from bot_core.abc.client_utils_abc import ClientUtilsABC
from bot_core.abc.permission_service_abc import PermissionServiceABC
from bot_data.abc.achievement_repository_abc import AchievementRepositoryABC
from bot_data.abc.user_joined_game_server_repository_abc import (
UserJoinedGameServerRepositoryABC,
@ -11,7 +12,6 @@ from bot_data.abc.user_joined_voice_channel_repository_abc import (
UserJoinedVoiceChannelRepositoryABC,
)
from bot_data.abc.user_warnings_repository_abc import UserWarningsRepositoryABC
from bot_data.model.user import User
from bot_data.model.user_history import UserHistory
from bot_graphql.abc.data_query_with_history_abc import DataQueryWithHistoryABC
from bot_graphql.filter.achievement_filter import AchievementFilter
@ -22,7 +22,6 @@ from bot_graphql.filter.user_joined_voice_channel_filter import (
)
from bot_graphql.filter.user_warning_filter import UserWarningFilter
from modules.level.service.level_service import LevelService
from modules.permission.abc.permission_service_abc import PermissionServiceABC
class UserQuery(DataQueryWithHistoryABC):
@ -50,15 +49,16 @@ class UserQuery(DataQueryWithHistoryABC):
self._permissions = permissions
self._achievements = achievements
self.set_field("id", self.resolve_id)
self.set_field("discordId", self.resolve_discord_id)
self.set_field("name", self.resolve_name)
self.set_field("xp", self.resolve_xp)
self.set_field("id", lambda user, *_: user.id)
self.set_field("discordId", lambda user, *_: user.discord_id)
self.set_field("name", lambda user, *_: user.name)
self.set_field("xp", lambda user, *_: user.xp)
self.set_field("messageCount", lambda x, *_: x.message_count)
self.set_field("reactionCount", lambda x, *_: x.reaction_count)
self.set_field("birthday", lambda x, *_: None if x.birthday is None else x.birthday.strftime("%d.%m.%Y"))
self.set_field("ontime", self.resolve_ontime)
self.set_field("level", self.resolve_level)
self.set_field("ontime", lambda user, *_: user.ontime)
self.set_field("gameOntime", lambda user, *_: user.game_ontime)
self.set_field("level", lambda user, *_: user.level)
self.add_collection(
"joinedServer",
lambda user, *_: self._ujs.get_user_joined_servers_by_user_id(user.id),
@ -85,37 +85,5 @@ class UserQuery(DataQueryWithHistoryABC):
UserWarningFilter,
)
self.set_field("server", self.resolve_server)
self.set_field("leftServer", self.resolve_left_server)
@staticmethod
def resolve_id(user: User, *_):
return user.id
@staticmethod
def resolve_discord_id(user: User, *_):
return user.discord_id
@staticmethod
def resolve_name(user: User, *_):
return user.name
@staticmethod
def resolve_xp(user: User, *_):
return user.xp
@staticmethod
def resolve_ontime(user: User, *_):
return user.ontime
@staticmethod
def resolve_level(user: User, *_):
return user.level
@staticmethod
def resolve_server(user: User, *_):
return user.server
@staticmethod
def resolve_left_server(user: User, *_):
return user.left_server
self.set_field("server", lambda user, *_: user.server)
self.set_field("leftServer", lambda user, *_: user.left_server)

14
bot/src/bot_graphql/query.py
View File

@ -131,6 +131,10 @@ class Query(QueryABC):
"hasFeatureFlag",
lambda *_, **kwargs: self._resolve_has_feature_flag(*_, **kwargs),
)
self.set_field(
"featureFlags",
lambda *_, **kwargs: self._resolve_feature_flags(*_),
)
def _resolve_has_feature_flag(self, *_, **kwargs):
settings: TechnicianConfig = self._config.get_configuration(TechnicianConfig)
@ -140,3 +144,13 @@ class Query(QueryABC):
"key": kwargs["flag"],
"value": FeatureFlagsSettings.get_flag_from_dict(settings.feature_flags, FeatureFlagsEnum(kwargs["flag"])),
}
def _resolve_feature_flags(self, *_) -> list[dict]:
settings: TechnicianConfig = self._config.get_configuration(TechnicianConfig)
return [
{
"key": flag,
"value": FeatureFlagsSettings.get_flag_from_dict(settings.feature_flags, FeatureFlagsEnum(flag)),
}
for flag in [e.value for e in FeatureFlagsEnum]
]

4
bot/src/modules/achievements/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "modules.achievements"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

2
bot/src/modules/achievements/achievements.json
View File

@ -4,7 +4,7 @@
"Version": {
"Major": "1",
"Minor": "2",
"Micro": "2"
"Micro": "3"
},
"Author": "Sven Heidemann",
"AuthorEmail": "sven.heidemann@sh-edraft.de",

4
bot/src/modules/achievements/commands/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "modules.achievements.commands"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

4
bot/src/modules/achievements/events/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "modules.achievements.events"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

4
bot/src/modules/achievements/model/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "modules.achievements.model"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

4
bot/src/modules/auto_role/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = "modules.auto_role"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2022 - 2023 sh-edraft.de"
__version__ = "1.2.2"
__version__ = "1.2.3"
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports:
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="1", minor="2", micro="2")
version_info = VersionInfo(major="1", minor="2", micro="3")

Some files were not shown because too many files have changed in this diff Show More