Fjerkroa/discord_bot
1
1
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Refactor discord_bot.py code a bit and improve comments

Browse Source
This commit is contained in:
OK
2023-04-13 16:09:00 +02:00
parent bcfe0e968f
commit b0f2f2f6e1
2 changed files with 80 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fjerkroa_bot/discord_bot.py
+46 -34
View File
@@ -27,39 +27,28 @@ class FjerkroaBot(commands.Bot):
intents.message_content = True
intents.members = True
self.observer = Observer()
self.file_handler = ConfigFileHandler(self.on_config_file_modified)
self.observer.schedule(self.file_handler, path=config_file, recursive=False)
self.observer.start()
self.airesponder = AIResponder(self.config)
self.aichannels = {}
for chan_name in self.config['additional-responders']:
self.aichannels[chan_name] = AIResponder(self.config, chan_name)
self.init_observer()
self.init_aichannels()
super().__init__(command_prefix="!", case_insensitive=True, intents=intents)
@classmethod
def load_config(self, config_file: str = "config.toml"):
with open(config_file, encoding='utf-8') as file:
return toml.load(file)
def init_observer(self):
self.observer = Observer()
self.file_handler = ConfigFileHandler(self.on_config_file_modified)
self.observer.schedule(self.file_handler, path=self.config_file, recursive=False)
self.observer.start()
def on_config_file_modified(self, event):
if event.src_path == self.config_file:
new_config = self.load_config(self.config_file)
if repr(new_config) != repr(self.config):
logging.info(f"config file {self.config_file} changed, reloading.")
self.config = new_config
self.airesponder.config = self.config
for responder in self.aichannels.values():
responder.config = self.config
def init_aichannels(self):
self.airesponder = AIResponder(self.config)
self.aichannels = {chan_name: AIResponder(self.config, chan_name) for chan_name in self.config['additional-responders']}
def init_channels(self):
self.staff_channel = self.fetch_channel_by_name(self.config['staff-channel'], no_ignore=True)
self.welcome_channel = self.fetch_channel_by_name(self.config['welcome-channel'], no_ignore=True)
async def on_ready(self):
print(f"We have logged in as {self.user}")
self.staff_channel = self.channel_by_name(self.config['staff-channel'], no_ignore=True)
logging.info(f'staff channel {repr(self.config.get("staff-channel"))}: {repr(self.staff_channel)}')
self.welcome_channel = self.channel_by_name(self.config['welcome-channel'], no_ignore=True)
logging.info(f'welcome channel {repr(self.config.get("welcome-channel"))}: {repr(self.welcome_channel)}')
self.init_channels()
logging.info(f"We have logged in as {self.user} ({repr(self.staff_channel)}, {repr(self.welcome_channel)})")
async def on_member_join(self, member):
logging.info(f"User {member.name} joined")
@@ -72,18 +61,29 @@ class FjerkroaBot(commands.Bot):
return
if not isinstance(message.channel, (TextChannel, DMChannel)):
return
message_content = str(message.content).strip()
if len(message_content) < 1:
return
channel_name = self.get_channel_name(message.channel)
msg = AIMessage(message.author.name, message_content, channel_name, self.user in message.mentions)
await self.respond(msg, message.channel)
await self.handle_message_through_responder(message)
def on_config_file_modified(self, event):
if event.src_path == self.config_file:
new_config = self.load_config(self.config_file)
if repr(new_config) != repr(self.config):
logging.info(f"config file {self.config_file} changed, reloading.")
self.config = new_config
self.airesponder.config = self.config
for responder in self.aichannels.values():
responder.config = self.config
@classmethod
def load_config(self, config_file: str = "config.toml"):
with open(config_file, encoding='utf-8') as file:
return toml.load(file)
def channel_by_name(self,
channel_name: Optional[str],
fallback_channel: Optional[Union[TextChannel, DMChannel]] = None,
no_ignore: bool = False
) -> Optional[Union[TextChannel, DMChannel]]:
"""Fetch a channel by name, or return the fallback channel if not found."""
if channel_name is None:
return fallback_channel
if channel_name.startswith("#"):
@@ -110,24 +110,36 @@ class FjerkroaBot(commands.Bot):
def get_ai_responder(self, channel_name):
return self.aichannels[channel_name] if channel_name in self.aichannels else self.airesponder
async def handle_message_through_responder(self, message):
"""Handle a message through the AI responder"""
message_content = str(message.content).strip()
if len(message_content) < 1:
return
channel_name = self.get_channel_name(message.channel)
msg = AIMessage(message.author.name, message_content, channel_name, self.user in message.mentions)
await self.respond(msg, message.channel)
async def send_message_with_typing(self, airesponder, channel, message):
"""Send the user message to the AI responder with typing animation in discord"""
async with channel.typing():
return await airesponder.send(message)
async def send_answer_with_typing(self, response, answer_channel, airesponder):
"""Send an answer from AI to discord channel with typing animation"""
async with answer_channel.typing():
if response.picture is not None:
# Generate the image with the AI and send it with the answer
images = [discord.File(fp=await airesponder.draw(response.picture), filename="image.png")]
await answer_channel.send(response.answer, files=images, suppress_embeds=True)
else:
await answer_channel.send(response.answer, suppress_embeds=True)
# This is an asynchronous function to generate AI responses
async def respond(
self,
message: AIMessage, # Incoming message object with user message and metadata
channel: Union[TextChannel, DMChannel] # Channel (Text or Direct Message) the message is coming from
) -> None:
"""Handle a message from a user with an AI responder"""
# Get the name of the channel where the message was sent
channel_name = self.get_channel_name(channel)