Ebola-Chan/kd_discord_bot
1
0
Fork 0
forked from sh-edraft.de/sh_discord_bot
Code Issues Pull Requests Projects Releases Wiki Activity

[WIP] Added basic commands & functionality of Gismo@0.4.10

Browse Source
This commit is contained in:
Sven Heidemann
2022-07-16 23:23:07 +02:00
parent f506cb74bb
commit 9f8dd52c82
95 changed files with 4385 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
src/bot_data/__init__.py
View File

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

26
src/bot_data/abc/__init__.py Normal file
View File

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

53
src/bot_data/abc/client_repository_abc.py Normal file
View File

@@ -0,0 +1,53 @@
from abc import ABC, abstractmethod
from typing import Optional
from cpl_query.extension import List
from bot_data.model.client import Client
class ClientRepositoryABC(ABC):
@abstractmethod
def __init__(self): pass
@abstractmethod
def get_clients(self) -> List[Client]: pass
@abstractmethod
def get_client_by_id(self, client_id: int) -> Client: pass
@abstractmethod
def get_client_by_discord_id(self, discord_id: int) -> Client: pass
@abstractmethod
def find_client_by_discord_id(self, discord_id: int) -> Optional[Client]: pass
@abstractmethod
def find_client_by_server_id(self, server_id: int) -> Optional[Client]: pass
@abstractmethod
def find_client_by_discord_id_and_server_id(self, discord_id: int, server_id: int) -> Optional[Client]: pass
@abstractmethod
def add_client(self, client: Client): pass
@abstractmethod
def update_client(self, client: Client): pass
@abstractmethod
def delete_client(self, client: Client): pass
@abstractmethod
def append_sent_message_count(self, client_id: int, server_id: int, value: int): pass
@abstractmethod
def append_received_message_count(self, client_id: int, server_id: int, value: int): pass
@abstractmethod
def append_deleted_message_count(self, client_id: int, server_id: int, value: int): pass
@abstractmethod
def append_received_command_count(self, client_id: int, server_id: int, value: int): pass
@abstractmethod
def append_moved_users_count(self, client_id: int, server_id: int, value: int): pass

30
src/bot_data/abc/known_user_repository_abc.py Normal file
View File

@@ -0,0 +1,30 @@
from abc import ABC, abstractmethod
from typing import Optional
from cpl_query.extension import List
from bot_data.model.known_user import KnownUser
class KnownUserRepositoryABC(ABC):
@abstractmethod
def __init__(self): pass
@abstractmethod
def get_users(self) -> List[KnownUser]: pass
@abstractmethod
def get_user_by_id(self, id: int) -> KnownUser: pass
@abstractmethod
def get_user_by_discord_id(self, discord_id: int) -> KnownUser: pass
@abstractmethod
def find_user_by_discord_id(self, discord_id: int) -> Optional[KnownUser]: pass
@abstractmethod
def add_user(self, known_user: KnownUser): pass
@abstractmethod
def delete_user(self, known_user: KnownUser): pass

13
src/bot_data/abc/migration_abc.py Normal file
View File

@@ -0,0 +1,13 @@
from abc import ABC, abstractmethod
class MigrationABC(ABC):
@abstractmethod
def __init__(self): pass
@abstractmethod
def upgrade(self): pass
@abstractmethod
def downgrade(self): pass

33
src/bot_data/abc/server_repository_abc.py Normal file
View File

@@ -0,0 +1,33 @@
from abc import ABC, abstractmethod
from typing import Optional
from cpl_query.extension import List
from bot_data.model.server import Server
class ServerRepositoryABC(ABC):
@abstractmethod
def __init__(self): pass
@abstractmethod
def get_servers(self) -> List[Server]: pass
@abstractmethod
def get_server_by_id(self, id: int) -> Server: pass
@abstractmethod
def get_server_by_discord_id(self, discord_id: int) -> Server: pass
@abstractmethod
def find_server_by_discord_id(self, discord_id: int) -> Optional[Server]: pass
@abstractmethod
def add_server(self, server: Server): pass
@abstractmethod
def update_server(self, server: Server): pass
@abstractmethod
def delete_server(self, server: Server): pass

35
src/bot_data/abc/user_joined_server_repository_abc.py Normal file
View File

@@ -0,0 +1,35 @@
from abc import ABC, abstractmethod
from typing import Optional
from cpl_query.extension import List
from bot_data.model.user_joined_server import UserJoinedServer
class UserJoinedServerRepositoryABC(ABC):
@abstractmethod
def __init__(self): pass
@abstractmethod
def get_user_joined_servers(self) -> List[UserJoinedServer]: pass
@abstractmethod
def get_user_joined_server_by_id(self, id: int) -> UserJoinedServer: pass
@abstractmethod
def get_user_joined_servers_by_user_id(self, user_id: int) -> list[UserJoinedServer]: pass
@abstractmethod
def get_active_user_joined_server_by_user_id(self, user_id: int) -> UserJoinedServer: pass
@abstractmethod
def find_active_user_joined_server_by_user_id(self, user_id: int) -> Optional[UserJoinedServer]: pass
@abstractmethod
def add_user_joined_server(self, user_joined_server: UserJoinedServer): pass
@abstractmethod
def update_user_joined_server(self, user_joined_server: UserJoinedServer): pass
@abstractmethod
def delete_user_joined_server(self, user_joined_server: UserJoinedServer): pass

37
src/bot_data/abc/user_joined_voice_channel_abc.py Normal file
View File

@@ -0,0 +1,37 @@
from abc import ABC, abstractmethod
from typing import Optional
from cpl_query.extension import List
from bot_data.model.user_joined_voice_channel import UserJoinedVoiceChannel
class UserJoinedVoiceChannelRepositoryABC(ABC):
@abstractmethod
def __init__(self): pass
@abstractmethod
def get_user_joined_voice_channels(self) -> List[UserJoinedVoiceChannel]: pass
@abstractmethod
def get_user_joined_voice_channel_by_id(self, id: int) -> UserJoinedVoiceChannel: pass
@abstractmethod
def get_user_joined_voice_channels_by_user_id(self, user_id: int) -> list[UserJoinedVoiceChannel]: pass
@abstractmethod
def get_active_user_joined_voice_channel_by_user_id(self, user_id: int) -> UserJoinedVoiceChannel: pass
@abstractmethod
def find_active_user_joined_voice_channel_by_user_id(self, user_id: int) -> Optional[UserJoinedVoiceChannel]: pass
@abstractmethod
def find_active_user_joined_voice_channels_by_user_id(self, user_id: int) -> List[Optional[UserJoinedVoiceChannel]]: pass
@abstractmethod
def add_user_joined_voice_channel(self, user_joined_voice_channel: UserJoinedVoiceChannel): pass
@abstractmethod
def update_user_joined_voice_channel(self, user_joined_voice_channel: UserJoinedVoiceChannel): pass
@abstractmethod
def delete_user_joined_voice_channel(self, user_joined_voice_channel: UserJoinedVoiceChannel): pass

36
src/bot_data/abc/user_repository_abc.py Normal file
View File

@@ -0,0 +1,36 @@
from abc import ABC, abstractmethod
from typing import Optional
from cpl_query.extension import List
from bot_data.model.user import User
class UserRepositoryABC(ABC):
@abstractmethod
def __init__(self): pass
@abstractmethod
def get_users(self) -> List[User]: pass
@abstractmethod
def get_user_by_id(self, id: int) -> User: pass
@abstractmethod
def get_users_by_discord_id(self, discord_id: int) -> List[User]: pass
@abstractmethod
def get_user_by_discord_id_and_server_id(self, discord_id: int, server_id: int) -> User: pass
@abstractmethod
def find_user_by_discord_id_and_server_id(self, discord_id: int, server_id: int) -> Optional[User]: pass
@abstractmethod
def add_user(self, user: User): pass
@abstractmethod
def update_user(self, user: User): pass
@abstractmethod
def delete_user(self, user: User): pass

33
src/bot_data/db_context.py Normal file
View File

@@ -0,0 +1,33 @@
from cpl_core.database import DatabaseSettings
from cpl_core.database.context import DatabaseContext
from cpl_core.logging import LoggerABC
class DBContext(DatabaseContext):
def __init__(self, logger: LoggerABC):
self._logger = logger
DatabaseContext.__init__(self)
def connect(self, database_settings: DatabaseSettings):
try:
self._logger.debug(__name__, "Connecting to database")
self._db.connect(database_settings)
self.save_changes()
self._logger.info(__name__, "Connected to database")
except Exception as e:
self._logger.fatal(__name__, "Connecting to database failed", e)
def save_changes(self):
try:
self._logger.trace(__name__, "Save changes")
super(DBContext, self).save_changes()
self._logger.debug(__name__, "Saved changes")
except Exception as e:
self._logger.error(__name__, "Saving changes failed", e)
def select(self, statement: str) -> list[tuple]:
return super(DBContext, self).select(statement)

26
src/bot_data/migration/__init__.py Normal file
View File

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

123
src/bot_data/migration/initial_migration.py Normal file
View File

@@ -0,0 +1,123 @@
from cpl_core.logging import LoggerABC
from bot_data.abc.migration_abc import MigrationABC
from bot_data.db_context import DBContext
class InitialMigration(MigrationABC):
def __init__(self, logger: LoggerABC, db: DBContext):
self._logger = logger
self._db = db
self._cursor = db.cursor
def upgrade(self):
self._logger.debug(__name__, 'Running upgrade')
self._cursor.execute(
str(f"""
CREATE TABLE IF NOT EXISTS `MigrationHistory` (
`MigrationId` VARCHAR(255),
`CreatedAt` DATETIME(6),
`LastModifiedAt` DATETIME(6),
PRIMARY KEY(`MigrationId`)
);
""")
)
self._cursor.execute(
str(f"""
CREATE TABLE IF NOT EXISTS `Servers` (
`ServerId` BIGINT NOT NULL AUTO_INCREMENT,
`DiscordServerId` BIGINT NOT NULL,
`CreatedAt` DATETIME(6),
`LastModifiedAt` DATETIME(6),
PRIMARY KEY(`ServerId`)
);
""")
)
self._cursor.execute(
str(f"""
CREATE TABLE IF NOT EXISTS `Users` (
`UserId` BIGINT NOT NULL AUTO_INCREMENT,
`DiscordId` BIGINT NOT NULL,
`XP` BIGINT NOT NULL DEFAULT 0,
`ServerId` BIGINT,
`CreatedAt` DATETIME(6),
`LastModifiedAt` DATETIME(6),
FOREIGN KEY (`ServerId`) REFERENCES Servers(`ServerId`),
PRIMARY KEY(`UserId`)
);
""")
)
self._cursor.execute(
str(f"""
CREATE TABLE IF NOT EXISTS `Clients` (
`ClientId` BIGINT NOT NULL AUTO_INCREMENT,
`DiscordClientId` BIGINT NOT NULL,
`SentMessageCount` BIGINT NOT NULL DEFAULT 0,
`ReceivedMessageCount` BIGINT NOT NULL DEFAULT 0,
`DeletedMessageCount` BIGINT NOT NULL DEFAULT 0,
`ReceivedCommandsCount` BIGINT NOT NULL DEFAULT 0,
`MovedUsersCount` BIGINT NOT NULL DEFAULT 0,
`ServerId` BIGINT,
`CreatedAt` DATETIME(6),
`LastModifiedAt` DATETIME(6),
FOREIGN KEY (`ServerId`) REFERENCES Servers(`ServerId`),
PRIMARY KEY(`ClientId`)
);
""")
)
self._cursor.execute(
str(f"""
CREATE TABLE IF NOT EXISTS `KnownUsers` (
`KnownUserId` BIGINT NOT NULL AUTO_INCREMENT,
`DiscordId` BIGINT NOT NULL,
`CreatedAt` DATETIME(6),
`LastModifiedAt` DATETIME(6),
PRIMARY KEY(`KnownUserId`)
);
""")
)
self._cursor.execute(
str(f"""
CREATE TABLE IF NOT EXISTS `UserJoinedServers` (
`JoinId` BIGINT NOT NULL AUTO_INCREMENT,
`UserId` BIGINT NOT NULL,
`JoinedOn` DATETIME(6) NOT NULL,
`LeavedOn` DATETIME(6),
`CreatedAt` DATETIME(6),
`LastModifiedAt` DATETIME(6),
FOREIGN KEY (`UserId`) REFERENCES Users(`UserId`),
PRIMARY KEY(`JoinId`)
);
""")
)
self._cursor.execute(
str(f"""
CREATE TABLE IF NOT EXISTS `UserJoinedVoiceChannel` (
`JoinId` BIGINT NOT NULL AUTO_INCREMENT,
`UserId` BIGINT NOT NULL,
`DiscordChannelId` BIGINT NOT NULL,
`JoinedOn` DATETIME(6) NOT NULL,
`LeavedOn` DATETIME(6),
`CreatedAt` DATETIME(6),
`LastModifiedAt` DATETIME(6),
FOREIGN KEY (`UserId`) REFERENCES Users(`UserId`),
PRIMARY KEY(`JoinId`)
);
""")
)
def downgrade(self):
self._cursor.execute('DROP TABLE `Servers`;')
self._cursor.execute('DROP TABLE `Users`;')
self._cursor.execute('DROP TABLE `Clients`;')
self._cursor.execute('DROP TABLE `KnownUsers`;')
self._cursor.execute('DROP TABLE `UserJoinedServers`;')
self._cursor.execute('DROP TABLE `UserJoinedVoiceChannel`;')

26
src/bot_data/model/__init__.py Normal file
View File

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

170
src/bot_data/model/client.py Normal file
View File

@@ -0,0 +1,170 @@
from datetime import datetime
from cpl_core.database import TableABC
from bot_data.model.server import Server
class Client(TableABC):
def __init__(self,
dc_id: int,
smc: int,
rmc: int,
dmc: int,
rcc: int,
muc: int,
server: Server,
created_at: datetime = None,
modified_at: datetime = None,
id=0
):
self._client_id = id
self._discord_client_id = dc_id
self._sent_message_count = smc
self._received_message_count = rmc
self._deleted_message_count = dmc
self._received_command_count = rcc
self._moved_users_count = muc
self._server: Server = server
TableABC.__init__(self)
self._created_at = created_at if created_at is not None else self._created_at
self._modified_at = modified_at if modified_at is not None else self._modified_at
@property
def client_id(self) -> int:
return self._client_id
@property
def discord_id(self) -> int:
return self._discord_client_id
@property
def sent_message_count(self) -> int:
return self._sent_message_count
@sent_message_count.setter
def sent_message_count(self, value: int):
self._modified_at = datetime.now().isoformat()
self._sent_message_count = value
@property
def received_message_count(self) -> int:
return self._received_message_count
@received_message_count.setter
def received_message_count(self, value: int):
self._modified_at = datetime.now().isoformat()
self._received_message_count = value
@property
def deleted_message_count(self) -> int:
return self._deleted_message_count
@deleted_message_count.setter
def deleted_message_count(self, value: int):
self._modified_at = datetime.now().isoformat()
self._deleted_message_count = value
@property
def received_command_count(self) -> int:
return self._received_command_count
@received_command_count.setter
def received_command_count(self, value: int):
self._modified_at = datetime.now().isoformat()
self._received_command_count = value
@property
def moved_users_count(self) -> int:
return self._moved_users_count
@moved_users_count.setter
def moved_users_count(self, value: int):
self._modified_at = datetime.now().isoformat()
self._moved_users_count = value
@property
def server(self) -> Server:
return self._server
@staticmethod
def get_select_all_string() -> str:
return str(f"""
SELECT * FROM `Clients`;
""")
@staticmethod
def get_select_by_id_string(id: int) -> str:
return str(f"""
SELECT * FROM `Clients`
WHERE `ClientId` = {id};
""")
@staticmethod
def get_select_by_discord_id_string(id: int) -> str:
return str(f"""
SELECT * FROM `Clients`
WHERE `DiscordClientId` = {id};
""")
@staticmethod
def get_select_by_server_id_string(id: int) -> str:
return str(f"""
SELECT * FROM `Clients`
WHERE `ServerId` = {id};
""")
@staticmethod
def get_select_by_discord_id_and_server_id_string(id: int, server_id: int) -> str:
return str(f"""
SELECT * FROM `Clients`
WHERE `DiscordClientId` = {id}
AND `ServerId` = {server_id};
""")
@property
def insert_string(self) -> str:
return str(f"""
INSERT INTO `Clients` (
`DiscordClientId`,
`SentMessageCount`,
`ReceivedMessageCount`,
`DeletedMessageCount`,
`ReceivedCommandsCount`,
`MovedUsersCount`,
`ServerId`,
`CreatedAt`,
`LastModifiedAt`
) VALUES (
{self._discord_client_id},
{self._sent_message_count},
{self._received_message_count},
{self._deleted_message_count},
{self._received_message_count},
{self._moved_users_count},
{self._server.server_id},
'{self._created_at}',
'{self._modified_at}'
);
""")
@property
def udpate_string(self) -> str:
return str(f"""
UPDATE `Clients`
SET `SentMessageCount` = {self._sent_message_count},
`ReceivedMessageCount` = {self._received_message_count},
`DeletedMessageCount` = {self._deleted_message_count},
`ReceivedCommandsCount` = {self._received_command_count},
`MovedUsersCount` = {self._moved_users_count},
`LastModifiedAt` = '{self._modified_at}'
WHERE `ClientId` = {self._client_id};
""")
@property
def delete_string(self) -> str:
return str(f"""
DELETE FROM `Clients`
WHERE `ClientId` = {self._client_id};
""")

67
src/bot_data/model/known_user.py Normal file
View File

@@ -0,0 +1,67 @@
from datetime import datetime
from typing import Optional
from cpl_core.database import TableABC
from bot_data.model.server import Server
class KnownUser(TableABC):
def __init__(self, dc_id: int, created_at: datetime = None, modified_at: datetime = None, id=0):
self._known_user_id = id
self._discord_id = dc_id
TableABC.__init__(self)
self._created_at = created_at if created_at is not None else self._created_at
self._modified_at = modified_at if modified_at is not None else self._modified_at
@property
def known_user_id(self) -> int:
return self._known_user_id
@property
def discord_id(self) -> int:
return self._discord_id
@staticmethod
def get_select_all_string() -> str:
return str(f"""
SELECT * FROM `KnownUsers`;
""")
@staticmethod
def get_select_by_id_string(id: int) -> str:
return str(f"""
SELECT * FROM `KnownUsers`
WHERE `KnownUserId` = {id};
""")
@staticmethod
def get_select_by_discord_id_string(id: int) -> str:
return str(f"""
SELECT * FROM `KnownUsers`
WHERE `DiscordId` = {id};
""")
@property
def insert_string(self) -> str:
return str(f"""
INSERT INTO `KnownUsers` (
`DiscordId`, `CreatedAt`, `LastModifiedAt`
) VALUES (
{self._discord_id},
'{self._created_at}',
'{self._modified_at}'
);
""")
@property
def udpate_string(self) -> str:
return ''
@property
def delete_string(self) -> str:
return str(f"""
DELETE FROM `KnownUsers`
WHERE `Id` = {self._known_user_id};
""")

47
src/bot_data/model/migration_history.py Normal file
View File

@@ -0,0 +1,47 @@
from cpl_core.database import TableABC
class MigrationHistory(TableABC):
def __init__(self, id: str):
self._id = id
TableABC.__init__(self)
@property
def migration_id(self) -> str:
return self._id
@staticmethod
def get_select_by_id_string(id: str) -> str:
return str(f"""
SELECT * FROM `MigrationHistory`
WHERE `MigrationId` = '{id}';
""")
@property
def insert_string(self) -> str:
return str(f"""
INSERT INTO `MigrationHistory` (
`MigrationId`, `CreatedAt`, `LastModifiedAt`
) VALUES (
'{self._id}',
'{self._created_at}',
'{self._modified_at}'
);
""")
@property
def udpate_string(self) -> str:
return str(f"""
UPDATE `MigrationHistory`
SET LastModifiedAt` = '{self._modified_at}'
WHERE `MigrationId` = '{self._id}';
""")
@property
def delete_string(self) -> str:
return str(f"""
DELETE FROM `MigrationHistory`
WHERE `MigrationId` = '{self._id}';
""")

71
src/bot_data/model/server.py Normal file
View File

@@ -0,0 +1,71 @@
from datetime import datetime
from typing import Optional
from cpl_core.database import TableABC
class Server(TableABC):
def __init__(self, dc_id: int, created_at: datetime=None, modified_at: datetime=None, id=0):
self._server_id = id
self._discord_server_id = dc_id
TableABC.__init__(self)
self._created_at = created_at if created_at is not None else self._created_at
self._modified_at = modified_at if modified_at is not None else self._modified_at
@property
def server_id(self) -> int:
return self._server_id
@property
def discord_server_id(self) -> int:
return self._discord_server_id
@staticmethod
def get_select_all_string() -> str:
return str(f"""
SELECT * FROM `Servers`;
""")
@staticmethod
def get_select_by_id_string(id: int) -> str:
return str(f"""
SELECT * FROM `Servers`
WHERE `ServerId` = {id};
""")
@staticmethod
def get_select_by_discord_id_string(id: int) -> str:
return str(f"""
SELECT * FROM `Servers`
WHERE `DiscordServerId` = {id};
""")
@property
def insert_string(self) -> str:
return str(f"""
INSERT INTO `Servers` (
`DiscordServerId`, `CreatedAt`, `LastModifiedAt`
) VALUES (
{self._discord_server_id},
'{self._created_at}',
'{self._modified_at}'
);
""")
@property
def udpate_string(self) -> str:
return str(f"""
UPDATE `Servers`
SET `DiscordServerId` = {self._discord_server_id},
`LastModifiedAt` = '{self._modified_at}'
WHERE `Id` = {self._server_id};
""")
@property
def delete_string(self) -> str:
return str(f"""
DELETE FROM `Servers`
WHERE `Id` = {self._server_id};
""")

97
src/bot_data/model/user.py Normal file
View File

@@ -0,0 +1,97 @@
from datetime import datetime
from typing import Optional
from cpl_core.database import TableABC
from bot_data.model.server import Server
class User(TableABC):
def __init__(self, dc_id: int, xp: int, server: Optional[Server], created_at: datetime = None, modified_at: datetime = None, id=0):
self._user_id = id
self._discord_id = dc_id
self._xp = xp
self._server = server
TableABC.__init__(self)
self._created_at = created_at if created_at is not None else self._created_at
self._modified_at = modified_at if modified_at is not None else self._modified_at
@property
def user_id(self) -> int:
return self._user_id
@property
def discord_id(self) -> int:
return self._discord_id
@property
def xp(self) -> int:
return self._xp
@xp.setter
def xp(self, value: int):
self._modified_at = datetime.now().isoformat()
self._xp = value
@property
def server(self) -> Optional[Server]:
return self._server
@staticmethod
def get_select_all_string() -> str:
return str(f"""
SELECT * FROM `Users`;
""")
@staticmethod
def get_select_by_id_string(id: int) -> str:
return str(f"""
SELECT * FROM `Users`
WHERE `UserId` = {id};
""")
@staticmethod
def get_select_by_discord_id_string(id: int) -> str:
return str(f"""
SELECT * FROM `Users`
WHERE `DiscordId` = {id};
""")
@staticmethod
def get_select_by_discord_id_and_server_id_string(dc_id: int, s_id: int) -> str:
return str(f"""
SELECT * FROM `Users`
WHERE `DiscordId` = {dc_id}
AND `ServerId` = {s_id};
""")
@property
def insert_string(self) -> str:
return str(f"""
INSERT INTO `Users` (
`DiscordId`, `XP`, `ServerId`, `CreatedAt`, `LastModifiedAt`
) VALUES (
{self._discord_id},
{self._xp},
{self._server.server_id},
'{self._created_at}',
'{self._modified_at}'
);
""")
@property
def udpate_string(self) -> str:
return str(f"""
UPDATE `Users`
SET `XP` = {self._xp},
`LastModifiedAt` = '{self._modified_at}'
WHERE `UserId` = {self._user_id};
""")
@property
def delete_string(self) -> str:
return str(f"""
DELETE FROM `Users`
WHERE `UserId` = {self._user_id};
""")

116
src/bot_data/model/user_joined_server.py Normal file
View File

@@ -0,0 +1,116 @@
from datetime import datetime
from typing import Optional
from cpl_core.database import TableABC
from bot_data.model.user import User
from bot_data.model.server import Server
class UserJoinedServer(TableABC):
def __init__(self, user: User, joined_on: datetime, leaved_on: datetime=None, created_at: datetime=None, modified_at: datetime=None, id=0):
self._join_id = id
self._user = user
self._joined_on = joined_on
self._leaved_on = leaved_on
TableABC.__init__(self)
self._created_at = created_at if created_at is not None else self._created_at
self._modified_at = modified_at if modified_at is not None else self._modified_at
@property
def join_id(self) -> int:
return self._join_id
@property
def user(self) -> User:
return self._user
@property
def joined_on(self) -> datetime:
return self._joined_on
@joined_on.setter
def joined_on(self, value: datetime):
self._modified_at = datetime.now()
self.joined_on = value
@property
def leaved_on(self) -> datetime:
return self._leaved_on
@leaved_on.setter
def leaved_on(self, value: datetime):
self._modified_at = datetime.now()
self._leaved_on = value
@staticmethod
def get_select_all_string() -> str:
return str(f"""
SELECT * FROM `UserJoinedServers`;
""")
@staticmethod
def get_select_by_id_string(id: int) -> str:
return str(f"""
SELECT * FROM `UserJoinedServers`
WHERE `JoinId` = {id};
""")
@staticmethod
def get_select_by_user_id_string(id: int) -> str:
return str(f"""
SELECT * FROM `UserJoinedServers`
WHERE `UserId` = {id};
""")
@staticmethod
def get_select_active_by_user_id_string(id: int) -> str:
return str(f"""
SELECT * FROM `UserJoinedServers`
WHERE `UserId` = {id}
AND `LeavedOn` IS NULL;
""")
@property
def insert_string(self) -> str:
if self._leaved_on is not None:
return str(f"""
INSERT INTO `UserJoinedServers` (
`UserId`, `JoinedOn`, `LeavedOn`, `CreatedAt`, `LastModifiedAt`
) VALUES (
{self._user.user_id},
'{self._joined_on}',
'{self._leaved_on}',
'{self._created_at}',
'{self._modified_at}'
);
""")
else:
return str(f"""
INSERT INTO `UserJoinedServers` (
`UserId`, `JoinedOn`, `CreatedAt`, `LastModifiedAt`
) VALUES (
{self._user.user_id},
'{self._joined_on}',
'{self._created_at}',
'{self._modified_at}'
);
""")
@property
def udpate_string(self) -> str:
return str(f"""
UPDATE `UserJoinedServers`
SET `LeavedOn` = '{self._leaved_on}',
`LastModifiedAt` = '{self._modified_at}'
WHERE `UserId` = {self._user.user_id};
""")
@property
def delete_string(self) -> str:
return str(f"""
DELETE FROM `UserJoinedServers`
WHERE `Id` = {self._join_id};
""")

121
src/bot_data/model/user_joined_voice_channel.py Normal file
View File

@@ -0,0 +1,121 @@
from datetime import datetime
from cpl_core.database import TableABC
from bot_data.model.user import User
class UserJoinedVoiceChannel(TableABC):
def __init__(self, user: User, dc_channel_id: int, joined_on: datetime, leaved_on: datetime = None, created_at: datetime = None, modified_at: datetime = None, id=0):
self._join_id = id
self._dc_channel_id = dc_channel_id
self._user = user
self._joined_on = joined_on
self._leaved_on = leaved_on
TableABC.__init__(self)
self._created_at = created_at if created_at is not None else self._created_at
self._modified_at = modified_at if modified_at is not None else self._modified_at
@property
def join_id(self) -> int:
return self._join_id
@property
def dc_channel_id(self) -> int:
return self._dc_channel_id
@property
def user(self) -> User:
return self._user
@property
def joined_on(self) -> datetime:
return self._joined_on
@joined_on.setter
def joined_on(self, value: datetime):
self._modified_at = datetime.now()
self.joined_on = value
@property
def leaved_on(self) -> datetime:
return self._leaved_on
@leaved_on.setter
def leaved_on(self, value: datetime):
self._modified_at = datetime.now()
self._leaved_on = value
@staticmethod
def get_select_all_string() -> str:
return str(f"""
SELECT * FROM `UserJoinedVoiceChannel`;
""")
@staticmethod
def get_select_by_id_string(id: int) -> str:
return str(f"""
SELECT * FROM `UserJoinedVoiceChannel`
WHERE `JoinId` = {id};
""")
@staticmethod
def get_select_by_user_id_string(id: int) -> str:
return str(f"""
SELECT * FROM `UserJoinedVoiceChannel`
WHERE `UserId` = {id};
""")
@staticmethod
def get_select_active_by_user_id_string(id: int) -> str:
return str(f"""
SELECT * FROM `UserJoinedVoiceChannel`
WHERE `UserId` = {id}
AND `LeavedOn` IS NULL;
""")
@property
def insert_string(self) -> str:
if self._leaved_on is not None:
return str(f"""
INSERT INTO `UserJoinedVoiceChannel` (
`UserId`, `DiscordChannelId`, `JoinedOn`, `LeavedOn`, `CreatedAt`, `LastModifiedAt`
) VALUES (
{self._user.user_id},
{self._dc_channel_id},
'{self._joined_on}',
'{self._leaved_on}',
'{self._created_at}',
'{self._modified_at}'
);
""")
else:
return str(f"""
INSERT INTO `UserJoinedVoiceChannel` (
`UserId`, `DiscordChannelId`, `JoinedOn`, `CreatedAt`, `LastModifiedAt`
) VALUES (
{self._user.user_id},
{self._dc_channel_id},
'{self._joined_on}',
'{self._created_at}',
'{self._modified_at}'
);
""")
@property
def udpate_string(self) -> str:
return str(f"""
UPDATE `UserJoinedVoiceChannel`
SET `LeavedOn` = '{self._leaved_on}',
`LastModifiedAt` = '{self._modified_at}'
WHERE `JoinId` = {self._join_id};
""")
@property
def delete_string(self) -> str:
return str(f"""
DELETE FROM `UserJoinedVoiceChannel`
WHERE `JoinId` = {self._join_id};
""")

26
src/bot_data/service/__init__.py Normal file
View File

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

173
src/bot_data/service/client_repository_service.py Normal file
View File

@@ -0,0 +1,173 @@
from typing import Optional
from cpl_core.database.context import DatabaseContextABC
from cpl_core.logging import LoggerABC
from cpl_query.extension import List
from bot_data.abc.client_repository_abc import ClientRepositoryABC
from bot_data.abc.server_repository_abc import ServerRepositoryABC
from bot_data.model.client import Client
class ClientRepositoryService(ClientRepositoryABC):
def __init__(self, logger: LoggerABC, db_context: DatabaseContextABC, servers: ServerRepositoryABC):
self._logger = logger
self._context = db_context
self._servers = servers
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]),
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))
return Client(
result[1],
result[2],
result[3],
result[4],
result[5],
result[6],
self._servers.get_server_by_id(result[7]),
id=result[0]
)
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]),
id=result[0]
)
def find_client_by_discord_id(self, discord_id: int) -> Optional[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))
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]),
id=result[0]
)
def find_client_by_server_id(self, discord_id: int) -> Optional[Client]:
self._logger.trace(__name__, f'Send SQL command: {Client.get_select_by_server_id_string(discord_id)}')
result = self._context.select(Client.get_select_by_server_id_string(discord_id))
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]),
id=result[0]
)
def find_client_by_discord_id_and_server_id(self, discord_id: int, server_id: int) -> Optional[Client]:
self._logger.trace(__name__, f'Send SQL command: {Client.get_select_by_discord_id_and_server_id_string(discord_id, server_id)}')
result = self._context.select(Client.get_select_by_discord_id_and_server_id_string(discord_id, server_id))
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]),
id=result[0]
)
def add_client(self, client: Client):
self._logger.trace(__name__, f'Send SQL command: {client.insert_string}')
self._context.cursor.execute(client.insert_string)
def update_client(self, client: Client):
self._logger.trace(__name__, f'Send SQL command: {client.udpate_string}')
self._context.cursor.execute(client.udpate_string)
def delete_client(self, client: Client):
self._logger.trace(__name__, f'Send SQL command: {client.delete_string}')
self._context.cursor.execute(client.delete_string)
def _get_client_and_server(self, id: int, server_id: int) -> Client:
server = self._servers.find_server_by_discord_id(server_id)
if server is None:
self._logger.warn(__name__, f'Cannot find server by id {server_id}')
raise Exception('Value not found')
client = self.find_client_by_discord_id_and_server_id(id, server.server_id)
if client is None:
self._logger.warn(__name__, f'Cannot find client by ids {id}@{server.server_id}')
raise Exception('Value not found')
return client
def append_sent_message_count(self, client_id: int, server_id: int, value: int):
client = self._get_client_and_server(client_id, server_id)
client.sent_message_count += value
self.update_client(client)
def append_received_message_count(self, client_id: int, server_id: int, value: int):
client = self._get_client_and_server(client_id, server_id)
client.received_message_count += value
self.update_client(client)
def append_deleted_message_count(self, client_id: int, server_id: int, value: int):
client = self._get_client_and_server(client_id, server_id)
client.deleted_message_count += value
self.update_client(client)
def append_received_command_count(self, client_id: int, server_id: int, value: int):
client = self._get_client_and_server(client_id, server_id)
client.received_command_count += value
self.update_client(client)
def append_moved_users_count(self, client_id: int, server_id: int, value: int):
client = self._get_client_and_server(client_id, server_id)
client.moved_users_count += value
self.update_client(client)

77
src/bot_data/service/known_user_repository_service.py Normal file
View File

@@ -0,0 +1,77 @@
from typing import Optional
from cpl_core.database.context import DatabaseContextABC
from cpl_core.logging import LoggerABC
from cpl_query.extension import List
from bot_data.abc.known_user_repository_abc import KnownUserRepositoryABC
from bot_data.abc.server_repository_abc import ServerRepositoryABC
from bot_data.model.known_user import KnownUser
class KnownUserRepositoryService(KnownUserRepositoryABC):
def __init__(self, logger: LoggerABC, db_context: DatabaseContextABC, servers: ServerRepositoryABC):
self._logger = logger
self._context = db_context
self._servers = servers
KnownUserRepositoryABC.__init__(self)
def get_users(self) -> List[KnownUser]:
users = List(KnownUser)
self._logger.trace(__name__, f'Send SQL command: {KnownUser.get_select_all_string()}')
results = self._context.select(KnownUser.get_select_all_string())
for result in results:
self._logger.trace(__name__, f'Get known_user with id {result[0]}')
users.append(KnownUser(
result[1],
result[2],
result[3],
id=result[0]
))
return users
def get_user_by_id(self, id: int) -> KnownUser:
self._logger.trace(__name__, f'Send SQL command: {KnownUser.get_select_by_id_string(id)}')
result = self._context.select(KnownUser.get_select_by_id_string(id))
return KnownUser(
result[1],
result[2],
result[3],
id=result[0]
)
def get_user_by_discord_id(self, discord_id: int) -> KnownUser:
self._logger.trace(__name__, f'Send SQL command: {KnownUser.get_select_by_discord_id_string(discord_id)}')
result = self._context.select(KnownUser.get_select_by_discord_id_string(discord_id))[0]
return KnownUser(
result[1],
result[2],
result[3],
id=result[0]
)
def find_user_by_discord_id(self, discord_id: int) -> Optional[KnownUser]:
self._logger.trace(__name__, f'Send SQL command: {KnownUser.get_select_by_discord_id_string(discord_id)}')
result = self._context.select(KnownUser.get_select_by_discord_id_string(discord_id))
if result is None or len(result) == 0:
return None
result = result[0]
return KnownUser(
result[1],
result[2],
result[3],
id=result[0]
)
def add_user(self, known_user: KnownUser):
self._logger.trace(__name__, f'Send SQL command: {known_user.insert_string}')
self._context.cursor.execute(known_user.insert_string)
def delete_user(self, known_user: KnownUser):
self._logger.trace(__name__, f'Send SQL command: {known_user.delete_string}')
self._context.cursor.execute(known_user.delete_string)

45
src/bot_data/service/migration_service.py Normal file
View File

@@ -0,0 +1,45 @@
from typing import Type
from cpl_core.database.context import DatabaseContextABC
from cpl_core.dependency_injection import ServiceProviderABC
from cpl_core.logging import LoggerABC
from bot_data.abc.migration_abc import MigrationABC
from bot_data.model.migration_history import MigrationHistory
class MigrationService:
def __init__(self, logger: LoggerABC, services: ServiceProviderABC, db: DatabaseContextABC):
self._logger = logger
self._services = services
self._db = db
self._cursor = db.cursor
self._migrations: list[Type[MigrationABC]] = MigrationABC.__subclasses__()
def migrate(self):
self._logger.info(__name__, f"Running Migrations")
for migration in self._migrations:
migration_id = migration.__name__
try:
# check if table exists
self._cursor.execute("SHOW TABLES LIKE 'MigrationHistory'")
result = self._cursor.fetchone()
if result:
# there is a table named "tableName"
self._logger.trace(__name__, f"Running SQL Command: {MigrationHistory.get_select_by_id_string(migration_id)}")
migration_from_db = self._db.select(MigrationHistory.get_select_by_id_string(migration_id))
self._logger.trace(__name__, str(migration_from_db))
if migration_from_db is not None and len(migration_from_db) > 0:
continue
self._logger.debug(__name__, f"Running Migration {migration}")
migration_as_service: MigrationABC = self._services.get_service(migration)
migration_as_service.upgrade()
self._cursor.execute(MigrationHistory(migration_id).insert_string)
self._db.save_changes()
except Exception as e:
self._logger.error(__name__, f'Cannot get migration with id {migration}', e)

71
src/bot_data/service/server_repository_service.py Normal file
View File

@@ -0,0 +1,71 @@
from typing import Optional
from cpl_core.database.context import DatabaseContextABC
from cpl_core.logging import LoggerABC
from cpl_query.extension import List
from bot_data.abc.server_repository_abc import ServerRepositoryABC
from bot_data.model.server import Server
class ServerRepositoryService(ServerRepositoryABC):
def __init__(self, logger: LoggerABC, db_context: DatabaseContextABC):
self._logger = logger
self._context = db_context
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],
id=result[0]
))
return servers
def get_server_by_id(self, server_id: int) -> Server:
self._logger.trace(__name__, f'Send SQL command: {Server.get_select_by_id_string(server_id)}')
result = self._context.select(Server.get_select_by_id_string(server_id))[0]
return Server(
result[1],
id=result[0]
)
def get_server_by_discord_id(self, discord_id: int) -> Server:
self._logger.trace(__name__, f'Send SQL command: {Server.get_select_by_discord_id_string(discord_id)}')
result = self._context.select(Server.get_select_by_discord_id_string(discord_id))[0]
return Server(
result[1],
id=result[0]
)
def find_server_by_discord_id(self, discord_id: int) -> Optional[Server]:
self._logger.trace(__name__, f'Send SQL command: {Server.get_select_by_discord_id_string(discord_id)}')
result = self._context.select(Server.get_select_by_discord_id_string(discord_id))
if result is None or len(result) == 0:
return None
result = result[0]
return Server(
result[1],
result[2],
result[3],
id=result[0]
)
def add_server(self, server: Server):
self._logger.trace(__name__, f'Send SQL command: {server.insert_string}')
self._context.cursor.execute(server.insert_string)
def update_server(self, server: Server):
self._logger.trace(__name__, f'Send SQL command: {server.udpate_string}')
self._context.cursor.execute(server.udpate_string)
def delete_server(self, server: Server):
self._logger.trace(__name__, f'Send SQL command: {server.delete_string}')
self._context.cursor.execute(server.delete_string)

107
src/bot_data/service/user_joined_server_repository_service.py Normal file
View File

@@ -0,0 +1,107 @@
from typing import Optional
from cpl_core.database.context import DatabaseContextABC
from cpl_core.logging import LoggerABC
from cpl_query.extension import List
from bot_data.abc.user_joined_server_repository_abc import \
UserJoinedServerRepositoryABC
from bot_data.abc.user_repository_abc import UserRepositoryABC
from bot_data.model.user import User
from bot_data.model.user_joined_server import UserJoinedServer
class UserJoinedServerRepositoryService(UserJoinedServerRepositoryABC):
def __init__(self, logger: LoggerABC, db_context: DatabaseContextABC, users: UserRepositoryABC):
self._logger = logger
self._context = db_context
self._users = users
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]
))
return joins
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_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
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]
)
def find_active_user_joined_server_by_user_id(self, user_id: int) -> Optional[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))
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]
)
def add_user_joined_server(self, user_joined_server: UserJoinedServer):
self._logger.trace(__name__, f'Send SQL command: {user_joined_server.insert_string}')
self._context.cursor.execute(user_joined_server.insert_string)
def update_user_joined_server(self, user_joined_server: UserJoinedServer):
self._logger.trace(__name__, f'Send SQL command: {user_joined_server.udpate_string}')
self._context.cursor.execute(user_joined_server.udpate_string)
def delete_user_joined_server(self, user_joined_server: UserJoinedServer):
self._logger.trace(__name__, f'Send SQL command: {user_joined_server.delete_string}')
self._context.cursor.execute(user_joined_server.delete_string)

123
src/bot_data/service/user_joined_voice_channel_service.py Normal file
View File

@@ -0,0 +1,123 @@
from typing import Optional
from cpl_core.database.context import DatabaseContextABC
from cpl_core.logging import LoggerABC
from cpl_query.extension import List, IterableABC
from bot_data.abc.user_repository_abc import UserRepositoryABC
from bot_data.model.user_joined_voice_channel import UserJoinedVoiceChannel
from bot_data.abc.user_joined_voice_channel_abc import UserJoinedVoiceChannelRepositoryABC
class UserJoinedVoiceChannelRepositoryService(UserJoinedVoiceChannelRepositoryABC):
def __init__(self, logger: LoggerABC, db_context: DatabaseContextABC, users: UserRepositoryABC):
self._logger = logger
self._context = db_context
self._users = users
UserJoinedVoiceChannelRepositoryABC.__init__(self)
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
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]
)
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
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]
)
def find_active_user_joined_voice_channel_by_user_id(self, user_id: int) -> Optional[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))
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]
)
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_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
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}')
self._context.cursor.execute(user_joined_voice_channel.insert_string)
def update_user_joined_voice_channel(self, user_joined_voice_channel: UserJoinedVoiceChannel):
self._logger.trace(__name__, f'Send SQL command: {user_joined_voice_channel.udpate_string}')
self._context.cursor.execute(user_joined_voice_channel.udpate_string)
def delete_user_joined_voice_channel(self, user_joined_voice_channel: UserJoinedVoiceChannel):
self._logger.trace(__name__, f'Send SQL command: {user_joined_voice_channel.delete_string}')
self._context.cursor.execute(user_joined_voice_channel.delete_string)

99
src/bot_data/service/user_repository_service.py Normal file
View File

@@ -0,0 +1,99 @@
from typing import Optional
from cpl_core.database.context import DatabaseContextABC
from cpl_core.logging import LoggerABC
from cpl_query.extension import List
from bot_data.abc.server_repository_abc import ServerRepositoryABC
from bot_data.abc.user_repository_abc import UserRepositoryABC
from bot_data.model.user import User
class UserRepositoryService(UserRepositoryABC):
def __init__(self, logger: LoggerABC, db_context: DatabaseContextABC, servers: ServerRepositoryABC):
self._logger = logger
self._context = db_context
self._servers = servers
UserRepositoryABC.__init__(self)
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(User(
result[1],
result[2],
self._servers.get_server_by_id(result[3]),
id=result[0]
))
return users
def get_user_by_id(self, id: int) -> User:
self._logger.trace(__name__, f'Send SQL command: {User.get_select_by_id_string(id)}')
result = self._context.select(User.get_select_by_id_string(id))[0]
return User(
result[1],
result[2],
self._servers.get_server_by_id(result[3]),
id=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(User(
result[1],
result[2],
self._servers.get_server_by_id(result[3]),
id=result[0]
))
return users
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 User(
result[1],
result[2],
self._servers.get_server_by_id(result[3]),
id=result[0]
)
def find_user_by_discord_id_and_server_id(self, discord_id: int, server_id: int) -> Optional[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))
if result is None or len(result) == 0:
return None
result = result[0]
return User(
result[1],
result[2],
self._servers.get_server_by_id(result[3]),
result[4],
result[5],
id=result[0]
)
def add_user(self, user: User):
self._logger.trace(__name__, f'Send SQL command: {user.insert_string}')
self._context.cursor.execute(user.insert_string)
def update_user(self, user: User):
self._logger.trace(__name__, f'Send SQL command: {user.udpate_string}')
self._context.cursor.execute(user.udpate_string)
def delete_user(self, user: User):
self._logger.trace(__name__, f'Send SQL command: {user.delete_string}')
self._context.cursor.execute(user.delete_string)