sh-edraft.de/sh_cpl
2
1
Fork 0
Code Issues 4 Pull Requests 1 Actions Packages Projects 1 Releases 12 Wiki Activity

Added logic to create a bot

Browse Source
This commit is contained in:
Sven Heidemann
2022-07-16 14:53:54 +02:00
parent 4402bcce73
commit fb35e8b8b2
44 changed files with 978 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
src/cpl_discord/__init__.py
View File

@@ -1 +1,33 @@
# imports
# build-ignore
def add_discord(self):
from cpl_core.console import Console
from cpl_discord.service.discord_bot_service_abc import DiscordBotServiceABC
from cpl_discord.service.discord_bot_service import DiscordBotService
from cpl_discord.service.discord_service_abc import DiscordServiceABC
from cpl_discord.service.discord_service import DiscordService
try:
self.add_singleton(DiscordServiceABC, DiscordService)
self.add_singleton(DiscordBotServiceABC, DiscordBotService)
except ImportError as e:
Console.error('cpl-discord is not installed', str(e))
def init():
from cpl_core.dependency_injection import ServiceCollection
ServiceCollection.add_discord = add_discord
init()
def get_discord_collection(services: 'ServiceCollectionABC') -> 'DiscordCollectionABC':
from cpl_discord.service.discord_collection import DiscordCollection
from cpl_discord.service.discord_collection_abc import DiscordCollectionABC
collection = DiscordCollection(services)
services.add_singleton(DiscordCollectionABC, collection)
return collection
# build-ignore-end

1
src/cpl_discord/application/__init__.py Normal file
View File

@@ -0,0 +1 @@
# imports

12
src/cpl_discord/application/discord_bot_abc.py Normal file
View File

@@ -0,0 +1,12 @@
from abc import abstractmethod
from cpl_core.application import ApplicationABC
class DiscordBotApplicationABC(ApplicationABC):
def __init__(self):
pass
@abstractmethod
def stop_async(self): pass

4
src/cpl_discord/command/command_abc.py → src/cpl_discord/command/discord_command_abc.py
View File

@@ -2,10 +2,10 @@ from abc import ABC, abstractmethod
from discord.ext import commands
from commands_meta import CommandsMeta
from discord_commands_meta import DiscordCogMeta
class CommandABC(ABC, commands.Cog, metaclass=CommandsMeta):
class DiscordCommandABC(ABC, commands.Cog, metaclass=DiscordCogMeta):
@abstractmethod
def __init__(self): pass

2
src/cpl_discord/command/commands_meta.py → src/cpl_discord/command/discord_commands_meta.py
View File

@@ -2,4 +2,4 @@ from abc import ABCMeta
from discord.ext import commands
class CommandsMeta(ABCMeta, commands.CogMeta): pass
class DiscordCogMeta(ABCMeta, commands.CogMeta): pass

2
src/cpl_discord/configuration/bot_settings.py → src/cpl_discord/configuration/discord_bot_settings.py
View File

@@ -4,7 +4,7 @@ from cpl_core.configuration.configuration_model_abc import ConfigurationModelABC
from cpl_core.console import Console
class BotSettings(ConfigurationModelABC):
class DiscordBotSettings(ConfigurationModelABC):
def __init__(self):
ConfigurationModelABC.__init__(self)

98
src/cpl_discord/event_types_enum.py Normal file
View File

@@ -0,0 +1,98 @@
from enum import Enum
from cpl_discord.events.on_bulk_message_delete_abc import OnBulkMessageDeleteABC
from cpl_discord.events.on_connect_abc import OnConnectABC
from cpl_discord.events.on_disconnect_abc import OnDisconnectABC
from cpl_discord.events.on_group_join_abc import OnGroupJoinABC
from cpl_discord.events.on_group_remove_abc import OnGroupRemoveABC
from cpl_discord.events.on_guild_available_abc import OnGuildAvailableABC
from cpl_discord.events.on_guild_channel_create_abc import OnGuildChannelCreateABC
from cpl_discord.events.on_guild_channel_delete_abc import OnGuildChannelDeleteABC
from cpl_discord.events.on_guild_channel_pins_update_abc import OnGuildChannelPinsUpdateABC
from cpl_discord.events.on_guild_channel_update_abc import OnGuildChannelUpdateABC
from cpl_discord.events.on_guild_emojis_update_abc import OnGuildEmojisUpdateABC
from cpl_discord.events.on_guild_integrations_update_abc import OnGuildIntegrationsUpdateABC
from cpl_discord.events.on_guild_join_abc import OnGuildJoinABC
from cpl_discord.events.on_guild_remove_abc import OnGuildRemoveABC
from cpl_discord.events.on_guild_role_create_abc import OnGuildRoleCreateABC
from cpl_discord.events.on_guild_role_delete_abc import OnGuildRoleDeleteABC
from cpl_discord.events.on_guild_role_update_abc import OnGuildRoleUpdateABC
from cpl_discord.events.on_guild_unavailable_abc import OnGuildUnavailableABC
from cpl_discord.events.on_guild_update_abc import OnGuildUpdateABC
from cpl_discord.events.on_invite_create_abc import OnInviteCreateABC
from cpl_discord.events.on_invite_delete_abc import OnInviteDeleteABC
from cpl_discord.events.on_member_ban_abc import OnMemberBanABC
from cpl_discord.events.on_member_join_abc import OnMemberJoinABC
from cpl_discord.events.on_member_remove_abc import OnMemberRemoveABC
from cpl_discord.events.on_member_unban_abc import OnMemberUnbanABC
from cpl_discord.events.on_member_update_abc import OnMemberUpdateABC
from cpl_discord.events.on_message_abc import OnMessageABC
from cpl_discord.events.on_message_delete_abc import OnMessageDeleteABC
from cpl_discord.events.on_message_edit_abc import OnMessageEditABC
from cpl_discord.events.on_private_channel_create_abc import OnPrivateChannelCreateABC
from cpl_discord.events.on_private_channel_delete_abc import OnPrivateChannelDeleteABC
from cpl_discord.events.on_private_channel_pins_update_abc import OnPrivateChannelPinsUpdateABC
from cpl_discord.events.on_private_channel_update_abc import OnPrivateChannelUpdateABC
from cpl_discord.events.on_reaction_add_abc import OnReactionAddABC
from cpl_discord.events.on_reaction_clear_abc import OnReactionClearABC
from cpl_discord.events.on_reaction_clear_emoji_abc import OnReactionClearEmojiABC
from cpl_discord.events.on_reaction_remove_abc import OnReactionRemoveABC
from cpl_discord.events.on_ready_abc import OnReadyABC
from cpl_discord.events.on_relationship_add_abc import OnRelationshipAddABC
from cpl_discord.events.on_relationship_remove_abc import OnRelationshipRemoveABC
from cpl_discord.events.on_relationship_update_abc import OnRelationshipUpdateABC
from cpl_discord.events.on_resume_abc import OnResumeABC
from cpl_discord.events.on_typing_abc import OnTypingABC
from cpl_discord.events.on_user_update_abc import OnUserUpdateABC
from cpl_discord.events.on_voice_state_update_abc import OnVoiceStateUpdateABC
from cpl_discord.events.on_webhooks_update_abc import OnWebhooksUpdateABC
class EventTypesEnum(Enum):
on_ready = OnReadyABC
on_bulk_message_delete_abc = OnBulkMessageDeleteABC
on_connect_abc = OnConnectABC
on_disconnect_abc = OnDisconnectABC
on_group_join_abc = OnGroupJoinABC
on_group_remove_abc = OnGroupRemoveABC
on_guild_available_abc = OnGuildAvailableABC
on_guild_channel_create_abc = OnGuildChannelCreateABC
on_guild_channel_delete_abc = OnGuildChannelDeleteABC
on_guild_channel_pins_update_abc = OnGuildChannelPinsUpdateABC
on_guild_channel_update_abc = OnGuildChannelUpdateABC
on_guild_emojis_update_abc = OnGuildEmojisUpdateABC
on_guild_integrations_update_abc = OnGuildIntegrationsUpdateABC
on_guild_join_abc = OnGuildJoinABC
on_guild_remove_abc = OnGuildRemoveABC
on_guild_role_create_abc = OnGuildRoleCreateABC
on_guild_role_delete_abc = OnGuildRoleDeleteABC
on_guild_role_update_abc = OnGuildRoleUpdateABC
on_guild_unavailable_abc = OnGuildUnavailableABC
on_guild_update_abc = OnGuildUpdateABC
on_invite_create_abc = OnInviteCreateABC
on_invite_delete_abc = OnInviteDeleteABC
on_member_ban_abc = OnMemberBanABC
on_member_join_abc = OnMemberJoinABC
on_member_remove_abc = OnMemberRemoveABC
on_member_unban_abc = OnMemberUnbanABC
on_member_update_abc = OnMemberUpdateABC
on_message_abc = OnMessageABC
on_message_delete_abc = OnMessageDeleteABC
on_message_edit_abc = OnMessageEditABC
on_private_channel_create_abc = OnPrivateChannelCreateABC
on_private_channel_delete_abc = OnPrivateChannelDeleteABC
on_private_channel_pins_update_abc = OnPrivateChannelPinsUpdateABC
on_private_channel_update_abc = OnPrivateChannelUpdateABC
on_reaction_add_abc = OnReactionAddABC
on_reaction_clear_abc = OnReactionClearABC
on_reaction_clear_emoji_abc = OnReactionClearEmojiABC
on_reaction_remove_abc = OnReactionRemoveABC
on_ready_abc = OnReadyABC
on_relationship_add_abc = OnRelationshipAddABC
on_relationship_remove_abc = OnRelationshipRemoveABC
on_relationship_update_abc = OnRelationshipUpdateABC
on_resume_abc = OnResumeABC
on_typing_abc = OnTypingABC
on_user_update_abc = OnUserUpdateABC
on_voice_state_update_abc = OnVoiceStateUpdateABC
on_webhooks_update_abc = OnWebhooksUpdateABC

71
src/cpl_discord/service/discord_bot_service.py Normal file
View File

@@ -0,0 +1,71 @@
import sys
import discord
from discord.ext import commands
from cpl_core.configuration import ConfigurationABC
from cpl_core.console import Console
from cpl_core.environment import ApplicationEnvironmentABC
from cpl_core.logging import LoggerABC, LoggingSettings, LoggingLevelEnum
from cpl_discord.configuration.discord_bot_settings import DiscordBotSettings
from cpl_discord.service.discord_bot_service_abc import DiscordBotServiceABC
from cpl_discord.service.discord_service_abc import DiscordServiceABC
class DiscordBotService(DiscordBotServiceABC):
def __init__(
self,
config: ConfigurationABC,
logger: LoggerABC,
discord_bot_settings: DiscordBotSettings,
env: ApplicationEnvironmentABC,
logging_st: LoggingSettings,
discord_service: DiscordServiceABC
):
# services
self._logger = logger
self._env = env
self._logging_st = logging_st
self._discord_service = discord_service
# settings
if discord_bot_settings is None:
self._discord_settings = DiscordBotSettings()
token = config.get_configuration('TOKEN')
if token is None:
raise Exception('You have to configure discord token by appsettings or environment variables')
prefix = config.get_configuration('PREFIX')
self._discord_settings.from_dict({
'Token': token,
'Prefix': prefix if prefix is not None else '! '
})
else:
self._discord_settings = discord_bot_settings
# setup super
DiscordBotServiceABC.__init__(self, command_prefix=self._discord_settings.prefix, help_command=None, intents=discord.Intents().all())
async def start_async(self):
self._logger.trace(__name__, 'Try to connect to discord')
await self.start(self._discord_settings.token)
# continue at on_ready
async def stop_async(self):
self._logger.trace(__name__, 'Try to disconnect from discord')
try:
await self.close()
except Exception as e:
self._logger.error(__name__, 'Stop failed', e)
async def on_ready(self):
self._logger.info(__name__, 'Connected to discord')
self._logger.header(f'{self.user.name}:')
if self._logging_st.console.value >= LoggingLevelEnum.INFO.value:
Console.banner(self._env.application_name if self._env.application_name != '' else 'A bot')
self.add_cog(self._discord_service)
await self._discord_service.on_ready()

9
src/cpl_discord/service/bot_service_abc.py → src/cpl_discord/service/discord_bot_service_abc.py
View File

@@ -4,13 +4,16 @@ import discord
from discord.ext import commands
class BotServiceABC(ABC, commands.Bot):
class DiscordBotServiceABC(commands.Bot):
def __init__(self):
ABC.__init__(self)
def __init__(self, **kwargs):
commands.Bot.__init__(self, **kwargs)
@abstractmethod
async def start_async(self): pass
@abstractmethod
async def stop_async(self): pass
@abstractmethod
async def on_ready(self): pass

32
src/cpl_discord/service/discord_collection.py Normal file
View File

@@ -0,0 +1,32 @@
from typing import Type, Optional
from cpl_core.console import Console
from cpl_core.dependency_injection import ServiceCollectionABC
from cpl_discord.service.discord_collection_abc import DiscordCollectionABC
from cpl_query.extension import List
from cpl_cli import CommandABC
class DiscordCollection(DiscordCollectionABC):
def __init__(self, service_collection: ServiceCollectionABC):
DiscordCollectionABC.__init__(self)
self._services = service_collection
self._events: dict[str, List] = {}
def add_command(self, _t: Type[CommandABC]):
self._services.add_transient(CommandABC, _t)
def add_event(self, _t_event: Type, _t: Type):
self._services.add_transient(_t_event, _t)
if _t_event not in self._events:
self._events[_t_event] = List(type(_t_event))
self._events[_t_event].append(_t)
def get_events_by_base(self, _t_event: Type) -> Optional[List]:
if _t_event not in self._events:
return None
return self._events[_t_event]

21
src/cpl_discord/service/discord_collection_abc.py Normal file
View File

@@ -0,0 +1,21 @@
from abc import ABC, abstractmethod
from typing import Type
from cpl_query.extension import List
from cpl_cli import CommandABC
class DiscordCollectionABC(ABC):
def __init__(self):
ABC.__init__(self)
@abstractmethod
def add_command(self, _t: Type[CommandABC]): pass
@abstractmethod
def add_event(self, _t_event: Type, _t: Type): pass
@abstractmethod
def get_events_by_base(self, _t_event: Type): pass

327
src/cpl_discord/service/discord_service.py Normal file
View File

@@ -0,0 +1,327 @@
from datetime import datetime
from typing import Optional, Sequence, Union, Type
import discord
from discord.ext import commands
from cpl_core.console import Console
from cpl_core.dependency_injection import ServiceProviderABC
from cpl_core.logging import LoggerABC
from cpl_core.utils import String
from cpl_discord.command.discord_commands_meta import DiscordCogMeta
from cpl_discord.events.on_bulk_message_delete_abc import OnBulkMessageDeleteABC
from cpl_discord.events.on_connect_abc import OnConnectABC
from cpl_discord.events.on_disconnect_abc import OnDisconnectABC
from cpl_discord.events.on_group_join_abc import OnGroupJoinABC
from cpl_discord.events.on_group_remove_abc import OnGroupRemoveABC
from cpl_discord.events.on_guild_available_abc import OnGuildAvailableABC
from cpl_discord.events.on_guild_channel_create_abc import OnGuildChannelCreateABC
from cpl_discord.events.on_guild_channel_delete_abc import OnGuildChannelDeleteABC
from cpl_discord.events.on_guild_channel_pins_update_abc import OnGuildChannelPinsUpdateABC
from cpl_discord.events.on_guild_channel_update_abc import OnGuildChannelUpdateABC
from cpl_discord.events.on_guild_emojis_update_abc import OnGuildEmojisUpdateABC
from cpl_discord.events.on_guild_integrations_update_abc import OnGuildIntegrationsUpdateABC
from cpl_discord.events.on_guild_join_abc import OnGuildJoinABC
from cpl_discord.events.on_guild_remove_abc import OnGuildRemoveABC
from cpl_discord.events.on_guild_role_create_abc import OnGuildRoleCreateABC
from cpl_discord.events.on_guild_role_delete_abc import OnGuildRoleDeleteABC
from cpl_discord.events.on_guild_role_update_abc import OnGuildRoleUpdateABC
from cpl_discord.events.on_guild_unavailable_abc import OnGuildUnavailableABC
from cpl_discord.events.on_guild_update_abc import OnGuildUpdateABC
from cpl_discord.events.on_invite_create_abc import OnInviteCreateABC
from cpl_discord.events.on_invite_delete_abc import OnInviteDeleteABC
from cpl_discord.events.on_member_ban_abc import OnMemberBanABC
from cpl_discord.events.on_member_join_abc import OnMemberJoinABC
from cpl_discord.events.on_member_remove_abc import OnMemberRemoveABC
from cpl_discord.events.on_member_unban_abc import OnMemberUnbanABC
from cpl_discord.events.on_member_update_abc import OnMemberUpdateABC
from cpl_discord.events.on_message_abc import OnMessageABC
from cpl_discord.events.on_message_delete_abc import OnMessageDeleteABC
from cpl_discord.events.on_message_edit_abc import OnMessageEditABC
from cpl_discord.events.on_private_channel_create_abc import OnPrivateChannelCreateABC
from cpl_discord.events.on_private_channel_delete_abc import OnPrivateChannelDeleteABC
from cpl_discord.events.on_private_channel_pins_update_abc import OnPrivateChannelPinsUpdateABC
from cpl_discord.events.on_private_channel_update_abc import OnPrivateChannelUpdateABC
from cpl_discord.events.on_reaction_add_abc import OnReactionAddABC
from cpl_discord.events.on_reaction_clear_abc import OnReactionClearABC
from cpl_discord.events.on_reaction_clear_emoji_abc import OnReactionClearEmojiABC
from cpl_discord.events.on_reaction_remove_abc import OnReactionRemoveABC
from cpl_discord.events.on_ready_abc import OnReadyABC
from cpl_discord.events.on_relationship_add_abc import OnRelationshipAddABC
from cpl_discord.events.on_relationship_remove_abc import OnRelationshipRemoveABC
from cpl_discord.events.on_relationship_update_abc import OnRelationshipUpdateABC
from cpl_discord.events.on_resume_abc import OnResumeABC
from cpl_discord.events.on_typing_abc import OnTypingABC
from cpl_discord.events.on_user_update_abc import OnUserUpdateABC
from cpl_discord.events.on_voice_state_update_abc import OnVoiceStateUpdateABC
from cpl_discord.events.on_webhooks_update_abc import OnWebhooksUpdateABC
from cpl_discord.service.discord_collection_abc import DiscordCollectionABC
from cpl_discord.service.discord_service_abc import DiscordServiceABC
class DiscordService(DiscordServiceABC, commands.Cog, metaclass=DiscordCogMeta):
def __init__(
self,
logger: LoggerABC,
dc_collection: DiscordCollectionABC,
services: ServiceProviderABC
):
DiscordServiceABC.__init__(self)
self._logger = logger
self._collection = dc_collection
self._services = services
async def _handle_event(self, event: Type, *args):
event_collection = self._collection.get_events_by_base(event)
if event_collection is None:
return
for event_type in event_collection:
event_instance = self._services.get_service(event_type)
func_name = event.__name__
if func_name.endswith('ABC'):
func_name = func_name.replace('ABC', '')
func_name = String.convert_to_snake_case(func_name)
try:
func = getattr(event_instance, func_name)
await func(*args)
except Exception as e:
self._logger.error(__name__, f'Cannot execute {func_name} of {event_instance.__name__}', e)
@commands.Cog.listener()
async def on_connect(self):
self._logger.trace(__name__, f'Received on_connect')
await self._handle_event(OnConnectABC)
@commands.Cog.listener()
async def on_disconnect(self):
self._logger.trace(__name__, f'Received on_disconnect')
await self._handle_event(OnDisconnectABC)
async def on_ready(self):
self._logger.trace(__name__, f'Received on_ready')
await self._handle_event(OnReadyABC)
@commands.Cog.listener()
async def on_resume(self):
self._logger.trace(__name__, f'Received on_resume')
await self._handle_event(OnResumeABC)
@commands.Cog.listener()
async def on_error(self, event: str, *args, **kwargs):
self._logger.trace(__name__, f'Received on_error:\n\t{event}\n\t{args}\n\t{kwargs}')
await self._handle_event(OnReadyABC, event, *args, **kwargs)
@commands.Cog.listener()
async def on_typing(self, channel: discord.abc.Messageable, user: Union[discord.User, discord.Member], when: datetime):
self._logger.trace(__name__, f'Received on_typing:\n\t{channel}\n\t{user}\n\t{when}')
await self._handle_event(OnTypingABC, channel, user, when)
@commands.Cog.listener()
async def on_message(self, message: discord.Message):
self._logger.trace(__name__, f'Received on_message:\n\t{message}\n\t{message.content}')
await self._handle_event(OnMessageABC, message)
@commands.Cog.listener()
async def on_message_delete(self, message: discord.Message):
self._logger.trace(__name__, f'Received on_message_delete:\n\t{message}\n\t{message.content}')
await self._handle_event(OnMessageDeleteABC, message)
@commands.Cog.listener()
async def on_bulk_message_delete(self, messages: list[discord.Message]):
self._logger.trace(__name__, f'Received on_bulk_message_delete:\n\t{len(messages)}')
await self._handle_event(OnBulkMessageDeleteABC, messages)
@commands.Cog.listener()
async def on_message_edit(self, before: discord.Message, after: discord.Message):
self._logger.trace(__name__, f'Received on_message_edit:\n\t{before}\n\t{after}')
await self._handle_event(OnMessageEditABC, before, after)
@commands.Cog.listener()
async def on_reaction_add(self, reaction: discord.Reaction, user: discord.User):
self._logger.trace(__name__, f'Received on_reaction_add:\n\t{reaction}\n\t{user}')
await self._handle_event(OnReactionAddABC, reaction, user)
@commands.Cog.listener()
async def on_reaction_remove(self, reaction: discord.Reaction, user: discord.User):
self._logger.trace(__name__, f'Received on_reaction_remove:\n\t{reaction}\n\t{user}')
await self._handle_event(OnReactionRemoveABC, reaction, user)
@commands.Cog.listener()
async def on_reaction_clear(self, message: discord.Message, reactions: list[discord.Reaction]):
self._logger.trace(__name__, f'Received on_reaction_reon_reaction_clearmove:\n\t{message}\n\t{len(reactions)}')
await self._handle_event(OnReactionClearABC, message, reactions)
@commands.Cog.listener()
async def on_reaction_clear_emoji(self, reaction: discord.Reaction):
self._logger.trace(__name__, f'Received on_reaction_clear_emoji:\n\t{reaction}')
await self._handle_event(OnReactionClearEmojiABC, reaction)
@commands.Cog.listener()
async def on_private_channel_delete(self, channel: discord.abc.PrivateChannel):
self._logger.trace(__name__, f'Received on_private_channel_delete:\n\t{channel}')
await self._handle_event(OnPrivateChannelDeleteABC, channel)
@commands.Cog.listener()
async def on_private_channel_create(self, channel: discord.abc.PrivateChannel):
self._logger.trace(__name__, f'Received on_private_channel_create:\n\t{channel}')
await self._handle_event(OnPrivateChannelCreateABC, channel)
@commands.Cog.listener()
async def on_private_channel_update(self, before: discord.GroupChannel, after: discord.GroupChannel):
self._logger.trace(__name__, f'Received on_private_channel_update:\n\t{before}\n\t{after}')
await self._handle_event(OnPrivateChannelUpdateABC, before, after)
@commands.Cog.listener()
async def on_private_channel_pins_update(self, channel: discord.abc.PrivateChannel, list_pin: Optional[datetime]):
self._logger.trace(__name__, f'Received on_private_channel_pins_update:\n\t{channel}\n\t{list_pin}')
await self._handle_event(OnPrivateChannelPinsUpdateABC, channel, list_pin)
@commands.Cog.listener()
async def on_guild_channel_delete(self, channel: discord.abc.GuildChannel):
self._logger.trace(__name__, f'Received on_guild_channel_delete:\n\t{channel}')
await self._handle_event(OnGuildChannelDeleteABC, channel)
@commands.Cog.listener()
async def on_guild_channel_create(self, channel: discord.abc.GuildChannel):
self._logger.trace(__name__, f'Received on_guild_channel_create:\n\t{channel}')
await self._handle_event(OnGuildChannelCreateABC, channel)
@commands.Cog.listener()
async def on_guild_channel_update(self, before: discord.abc.GuildChannel, after: discord.abc.GuildChannel):
self._logger.trace(__name__, f'Received on_guild_channel_update:\n\t{before}\n\t{after}')
await self._handle_event(OnGuildChannelUpdateABC, before, after)
@commands.Cog.listener()
async def on_guild_channel_pins_update(self, channel: discord.abc.GuildChannel, list_pin: Optional[datetime]):
self._logger.trace(__name__, f'Received on_guild_channel_pins_update:\n\t{channel}\n\t{list_pin}')
await self._handle_event(OnGuildChannelPinsUpdateABC, channel, list_pin)
@commands.Cog.listener()
async def on_guild_integrations_update(self, guild: discord.Guild):
self._logger.trace(__name__, f'Received on_guild_integrations_update:\n\t{guild}')
await self._handle_event(OnGuildIntegrationsUpdateABC, guild)
@commands.Cog.listener()
async def on_webhooks_update(self, channel: discord.abc.GuildChannel):
self._logger.trace(__name__, f'Received on_webhooks_update:\n\t{channel}')
await self._handle_event(OnWebhooksUpdateABC, channel)
@commands.Cog.listener()
async def on_member_join(self, member: discord.Member):
self._logger.trace(__name__, f'Received on_member_join:\n\t{member}')
await self._handle_event(OnMemberJoinABC, member)
@commands.Cog.listener()
async def on_member_remove(self, member: discord.Member):
self._logger.trace(__name__, f'Received on_member_remove:\n\t{member}')
await self._handle_event(OnMemberRemoveABC, member)
@commands.Cog.listener()
async def on_member_update(self, before: discord.Member, after: discord.Member):
self._logger.trace(__name__, f'Received on_member_update:\n\t{before}\n\t{after}')
await self._handle_event(OnMemberUpdateABC, before, after)
@commands.Cog.listener()
async def on_user_update(self, before: discord.User, after: discord.User):
self._logger.trace(__name__, f'Received on_user_update:\n\t{before}\n\t{after}')
await self._handle_event(OnUserUpdateABC, before, after)
@commands.Cog.listener()
async def on_guild_join(self, guild: discord.Guild):
self._logger.trace(__name__, f'Received on_guild_join:\n\t{guild}')
await self._handle_event(OnGuildJoinABC, guild)
@commands.Cog.listener()
async def on_guild_remove(self, guild: discord.Guild):
self._logger.trace(__name__, f'Received on_guild_remove:\n\t{guild}')
await self._handle_event(OnGuildRemoveABC, guild)
@commands.Cog.listener()
async def on_guild_update(self, before: discord.Guild, after: discord.Guild):
self._logger.trace(__name__, f'Received on_guild_update:\n\t{before}\n\t{after}')
await self._handle_event(OnGuildUpdateABC, before, after)
@commands.Cog.listener()
async def on_guild_role_create(self, role: discord.Role):
self._logger.trace(__name__, f'Received on_guild_role_create:\n\t{role}')
await self._handle_event(OnGuildRoleCreateABC, role)
@commands.Cog.listener()
async def on_guild_role_delete(self, role: discord.Role):
self._logger.trace(__name__, f'Received on_guild_role_delete:\n\t{role}')
await self._handle_event(OnGuildRoleDeleteABC, role)
@commands.Cog.listener()
async def on_guild_role_update(self, before: discord.Role, after: discord.Role):
self._logger.trace(__name__, f'Received on_guild_role_update:\n\t{before}\n\t{after}')
await self._handle_event(OnGuildRoleUpdateABC, before, after)
@commands.Cog.listener()
async def on_guild_emojis_update(self, guild: discord.Guild, before: Sequence[discord.Emoji], after: Sequence[discord.Emoji]):
self._logger.trace(__name__, f'Received on_guild_emojis_update:\n\t{guild}\n\t{before}\n\t{after}')
await self._handle_event(OnGuildEmojisUpdateABC, guild, before, after)
@commands.Cog.listener()
async def on_guild_available(self, guild: discord.Guild):
self._logger.trace(__name__, f'Received on_guild_available:\n\t{guild}')
await self._handle_event(OnGuildAvailableABC, guild)
@commands.Cog.listener()
async def on_guild_unavailable(self, guild: discord.Guild):
self._logger.trace(__name__, f'Received on_guild_unavailable:\n\t{guild}')
await self._handle_event(OnGuildUnavailableABC, guild)
@commands.Cog.listener()
async def on_voice_state_update(self, member: discord.Member, before: discord.VoiceState, after: discord.VoiceState):
self._logger.trace(__name__, f'Received on_voice_state_update:\n\t{member}\n\t{before}\n\t{after}')
await self._handle_event(OnVoiceStateUpdateABC, member, before, after)
@commands.Cog.listener()
async def on_member_ban(self, guild: discord.Guild, user: discord.User):
self._logger.trace(__name__, f'Received on_member_ban:\n\t{guild}\n\t{user}')
await self._handle_event(OnMemberBanABC, guild, user)
@commands.Cog.listener()
async def on_member_unban(self, guild: discord.Guild, user: discord.User):
self._logger.trace(__name__, f'Received on_member_unban:\n\t{guild}\n\t{user}')
await self._handle_event(OnMemberUnbanABC, guild, user)
@commands.Cog.listener()
async def on_invite_create(self, invite: discord.Invite):
self._logger.trace(__name__, f'Received on_invite_create:\n\t{invite}')
await self._handle_event(OnInviteCreateABC, invite)
@commands.Cog.listener()
async def on_invite_delete(self, invite: discord.Invite):
self._logger.trace(__name__, f'Received on_invite_create:\n\t{invite}')
await self._handle_event(OnInviteDeleteABC, invite)
@commands.Cog.listener()
async def on_group_join(self, channel: discord.GroupChannel, user: discord.User):
self._logger.trace(__name__, f'Received on_group_join:\n\t{channel}\n\t{user}')
await self._handle_event(OnGroupJoinABC, channel, user)
@commands.Cog.listener()
async def on_group_remove(self, channel: discord.GroupChannel, user: discord.User):
self._logger.trace(__name__, f'Received on_group_remove:\n\t{channel}\n\t{user}')
await self._handle_event(OnGroupRemoveABC, channel, user)
@commands.Cog.listener()
async def on_relationship_add(self, relationship: discord.Relationship):
self._logger.trace(__name__, f'Received on_relationship_add:\n\t{relationship}')
await self._handle_event(OnRelationshipAddABC, relationship)
@commands.Cog.listener()
async def on_relationship_remove(self, relationship: discord.Relationship):
self._logger.trace(__name__, f'Received on_relationship_remove:\n\t{relationship}')
await self._handle_event(OnRelationshipRemoveABC, relationship)
@commands.Cog.listener()
async def on_relationship_update(self, before: discord.Relationship, after: discord.Relationship):
self._logger.trace(__name__, f'Received on_relationship_update:\n\t{before}\n\t{after}')
await self._handle_event(OnRelationshipUpdateABC, before, after)

152
src/cpl_discord/service/discord_service_abc.py Normal file
View File

@@ -0,0 +1,152 @@
from abc import ABC, abstractmethod
from datetime import datetime
from typing import Optional, Sequence, Union
import discord
class DiscordServiceABC(ABC):
def __init__(self):
ABC.__init__(self)
@abstractmethod
async def on_connect(self): pass
@abstractmethod
async def on_disconnect(self): pass
@abstractmethod
async def on_ready(self): pass
@abstractmethod
async def on_resume(self): pass
@abstractmethod
async def on_error(self, event: str, *args, **kwargs): pass
@abstractmethod
async def on_typing(self, channel: discord.abc.Messageable, user: Union[discord.User, discord.Member], when: datetime): pass
@abstractmethod
async def on_message(self, message: discord.Message): pass
@abstractmethod
async def on_message_delete(self, message: discord.Message): pass
@abstractmethod
async def on_bulk_message_delete(self, messages: list[discord.Message]): pass
@abstractmethod
async def on_message_edit(self, before: discord.Message, after: discord.Message): pass
@abstractmethod
async def on_reaction_add(self, reaction: discord.Reaction, user: discord.User): pass
@abstractmethod
async def on_reaction_remove(self, reaction: discord.Reaction, user: discord.User): pass
@abstractmethod
async def on_reaction_clear(self, message: discord.Message, reactions: list[discord.Reaction]): pass
@abstractmethod
async def on_reaction_clear_emoji(self, reaction: discord.Reaction): pass
@abstractmethod
async def on_private_channel_delete(self, channel: discord.abc.PrivateChannel): pass
@abstractmethod
async def on_private_channel_create(self, channel: discord.abc.PrivateChannel): pass
@abstractmethod
async def on_private_channel_update(self, before: discord.GroupChannel, after: discord.GroupChannel): pass
@abstractmethod
async def on_private_channel_pins_update(self, channel: discord.abc.PrivateChannel, list_pin: Optional[datetime]): pass
@abstractmethod
async def on_guild_channel_delete(self, channel: discord.abc.GuildChannel): pass
@abstractmethod
async def on_guild_channel_create(self, channel: discord.abc.GuildChannel): pass
@abstractmethod
async def on_guild_channel_update(self, before: discord.abc.GuildChannel, after: discord.abc.GuildChannel): pass
@abstractmethod
async def on_guild_channel_pins_update(self, channel: discord.abc.GuildChannel, list_pin: Optional[datetime]): pass
@abstractmethod
async def on_guild_integrations_update(self, guild: discord.Guild): pass
@abstractmethod
async def on_webhooks_update(self, channel: discord.abc.GuildChannel): pass
@abstractmethod
async def on_member_join(self, member: discord.Member): pass
@abstractmethod
async def on_member_remove(self, member: discord.Member): pass
@abstractmethod
async def on_member_update(self, before: discord.Member, after: discord.Member): pass
@abstractmethod
async def on_user_update(self, before: discord.User, after: discord.User): pass
@abstractmethod
async def on_guild_join(self, guild: discord.Guild): pass
@abstractmethod
async def on_guild_remove(self, guild: discord.Guild): pass
@abstractmethod
async def on_guild_update(self, before: discord.Guild, after: discord.Guild): pass
@abstractmethod
async def on_guild_role_create(self, role: discord.Role): pass
@abstractmethod
async def on_guild_role_delete(self, role: discord.Role): pass
@abstractmethod
async def on_guild_role_update(self, before: discord.Role, after: discord.Role): pass
@abstractmethod
async def on_guild_emojis_update(self, guild: discord.Guild, before: Sequence[discord.Emoji], after: Sequence[discord.Emoji]): pass
@abstractmethod
async def on_guild_available(self, guild: discord.Guild): pass
@abstractmethod
async def on_guild_unavailable(self, guild: discord.Guild): pass
@abstractmethod
async def on_voice_state_update(self, member: discord.Member, before: discord.VoiceState, after: discord.VoiceState): pass
@abstractmethod
async def on_member_ban(self, guild: discord.Guild, user: discord.User): pass
@abstractmethod
async def on_member_unban(self, guild: discord.Guild, user: discord.User): pass
@abstractmethod
async def on_invite_create(self, invite: discord.Invite): pass
@abstractmethod
async def on_invite_delete(self, invite: discord.Invite): pass
@abstractmethod
async def on_group_join(self, chhanel: discord.GroupChannel, user: discord.User): pass
@abstractmethod
async def on_group_remove(self, chhanel: discord.GroupChannel, user: discord.User): pass
@abstractmethod
async def on_relationship_add(self, relationship: discord.Relationship): pass
@abstractmethod
async def on_relationship_remove(self, relationship: discord.Relationship): pass
@abstractmethod
async def on_relationship_update(self, before: discord.Relationship, after: discord.Relationship): pass