Refactor message handling by introducing TextBot class; enhance reply and reaction logic
This commit is contained in:
parent
be0567ff0f
commit
bab22747b4
GPG key ID:
9833D3C5A25BD227
4 changed files with 177 additions and 83 deletions
|
|
@ -6,6 +6,7 @@ import tomllib
|
|||
import discord
|
||||
|
||||
from botbotbot.ai import AIBot
|
||||
from botbotbot.text import TextBot
|
||||
from botbotbot.tts import CambAI
|
||||
|
||||
|
||||
|
|
@ -50,81 +51,15 @@ def main() -> None:
|
|||
|
||||
bot = discord.Bot(description=description, intents=intents)
|
||||
|
||||
text_bot = TextBot(bot, aibot, word_list)
|
||||
text_bot.init_events()
|
||||
|
||||
shuffle_tasks = set()
|
||||
|
||||
@bot.listen("on_ready")
|
||||
async def on_ready() -> None:
|
||||
logger.info(f"We have logged in as {bot.user}")
|
||||
|
||||
async def reply(message: discord.Message) -> None:
|
||||
logger.info(f"Reply to {message.author}")
|
||||
mention = random.choices(
|
||||
[f"<@{message.author.id}>", "@everyone", "@here"], weights=(98, 1, 1)
|
||||
)[0]
|
||||
content = random.choice(
|
||||
(
|
||||
f"{mention}, {random.choice(word_list)}",
|
||||
f"{random.choice(word_list)}",
|
||||
)
|
||||
)
|
||||
|
||||
if isinstance(message.channel, discord.TextChannel) and random.random() < 0.1:
|
||||
await send_as_webhook(
|
||||
message.channel,
|
||||
message.author.display_name,
|
||||
message.author.avatar.url if message.author.avatar else None,
|
||||
content,
|
||||
)
|
||||
else:
|
||||
fct = random.choice((message.reply, message.channel.send))
|
||||
|
||||
await fct(content)
|
||||
|
||||
async def ai_reply(message: discord.Message) -> None:
|
||||
if aibot is None:
|
||||
return
|
||||
|
||||
logger.info(f"AI Reply to {message.author}")
|
||||
prompt = message.clean_content
|
||||
if prompt == "" and message.embeds and message.embeds[0].description:
|
||||
prompt = message.embeds[0].description
|
||||
|
||||
answer = aibot.answer(prompt)
|
||||
if not isinstance(answer, str):
|
||||
return
|
||||
|
||||
if len(answer) > 2000:
|
||||
embed = discord.Embed(
|
||||
description=answer,
|
||||
thumbnail="https://mistral.ai/images/favicon/favicon-32x32.png",
|
||||
)
|
||||
await message.reply(embed=embed)
|
||||
else:
|
||||
await message.reply(answer)
|
||||
|
||||
async def react(message: discord.Message) -> None:
|
||||
if message.guild is None:
|
||||
return
|
||||
await message.add_reaction(random.choice(message.guild.emojis))
|
||||
|
||||
@bot.listen("on_message")
|
||||
async def on_message(message: discord.Message) -> None:
|
||||
if message.flags.ephemeral:
|
||||
return
|
||||
|
||||
if message.author != bot.user and bot.user in message.mentions:
|
||||
logger.info(f"{message.author} metion")
|
||||
await random.choice((reply, react))(message)
|
||||
return
|
||||
|
||||
probas = [10, 5, 10]
|
||||
|
||||
func = random.choices(
|
||||
(reply, ai_reply, react, None), weights=probas + [100 - sum(probas)]
|
||||
)[0]
|
||||
if func is not None:
|
||||
await func(message)
|
||||
|
||||
@bot.listen("on_reaction_add")
|
||||
async def add_more_reaction(
|
||||
reaction: discord.Reaction, user: discord.Member | discord.User
|
||||
|
|
@ -329,18 +264,4 @@ def main() -> None:
|
|||
await ctx.respond(embed=embed)
|
||||
logger.info("FIN CHAN")
|
||||
|
||||
async def send_as_webhook(
|
||||
channel: discord.TextChannel,
|
||||
name: str,
|
||||
avatar_url: str | None,
|
||||
content: str,
|
||||
) -> None:
|
||||
webhooks = await channel.webhooks()
|
||||
webhook = discord.utils.get(webhooks, name="BotbotbotHook")
|
||||
|
||||
if webhook is None:
|
||||
webhook = await channel.create_webhook(name="BotbotbotHook")
|
||||
|
||||
await webhook.send(content=content, username=name, avatar_url=avatar_url)
|
||||
|
||||
bot.run(config.get("token"))
|
||||
|
|
|
|||
146
botbotbot/text.py
Normal file
146
botbotbot/text.py
Normal file
|
|
@ -0,0 +1,146 @@
|
|||
import logging
|
||||
import random
|
||||
from typing import Any, Callable, Coroutine
|
||||
|
||||
import discord
|
||||
import emoji
|
||||
|
||||
from botbotbot.ai import AIBot
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class TextBot:
|
||||
uni_emojis = tuple(k for k in emoji.EMOJI_DATA.keys() if len(k) == 1)
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
bot: discord.Bot,
|
||||
aibot: AIBot | None = None,
|
||||
wordlist: list[str] = [],
|
||||
rnd_weights: list[float] = [10, 5, 10],
|
||||
) -> None:
|
||||
self.bot = bot
|
||||
self.aibot = aibot
|
||||
self.word_list = wordlist
|
||||
self._rnd_weights = rnd_weights
|
||||
|
||||
def init_events(self) -> None:
|
||||
self.bot.add_listener(self.on_message, "on_message")
|
||||
|
||||
@property
|
||||
def rnd_weights(self) -> list[float]:
|
||||
return self._rnd_weights + [100 - sum(self._rnd_weights)]
|
||||
|
||||
@property
|
||||
def rnd_functions(
|
||||
self,
|
||||
) -> list[Callable[[discord.Message], Coroutine[Any, Any, None]]]:
|
||||
return [self.reply, self.ai_reply, self.react]
|
||||
|
||||
@property
|
||||
def rnd_functions_or_not(
|
||||
self,
|
||||
) -> list[Callable[[discord.Message], Coroutine[Any, Any, None]] | None]:
|
||||
return [*self.rnd_functions, None]
|
||||
|
||||
async def on_message(self, message: discord.Message) -> None:
|
||||
logger.debug(
|
||||
f"Received message from <{message.author}> on channel <{message.channel}>."
|
||||
)
|
||||
if message.flags.ephemeral:
|
||||
logger.debug("Ephemeral message, ignoring.")
|
||||
return
|
||||
|
||||
if message.author != self.bot.user and self.bot.user in message.mentions:
|
||||
logger.info(
|
||||
f"Mention from <{message.author}> in channel <{message.channel}>."
|
||||
)
|
||||
await random.choices(self.rnd_functions, weights=self.rnd_weights[:-1])[0](
|
||||
message
|
||||
)
|
||||
return
|
||||
|
||||
func = random.choices(self.rnd_functions_or_not, weights=self.rnd_weights)[0]
|
||||
|
||||
if func is None:
|
||||
logger.debug("No action.")
|
||||
return
|
||||
|
||||
await func(message)
|
||||
|
||||
async def reply(self, message: discord.Message) -> None:
|
||||
logger.info(f"Replying to <{message.author}> in channel <{message.channel}>.")
|
||||
|
||||
mention = random.choices(
|
||||
[f"<@{message.author.id}>", "@everyone", "@here"], weights=(97, 1, 2)
|
||||
)[0]
|
||||
content = random.choice(
|
||||
(
|
||||
f"{mention}, {random.choice(self.word_list)}",
|
||||
f"{random.choice(self.word_list)}",
|
||||
)
|
||||
)
|
||||
|
||||
if (
|
||||
isinstance(message.channel, discord.TextChannel)
|
||||
and random.random() < 10 / 100
|
||||
):
|
||||
await self.send_as_webhook(
|
||||
message.channel,
|
||||
message.author.display_name,
|
||||
message.author.avatar.url if message.author.avatar else None,
|
||||
content,
|
||||
)
|
||||
else:
|
||||
fct = random.choice((message.reply, message.channel.send))
|
||||
|
||||
await fct(content)
|
||||
|
||||
async def ai_reply(self, message: discord.Message) -> None:
|
||||
if self.aibot is None:
|
||||
logger.debug("No AI bot, ignoring.")
|
||||
return
|
||||
|
||||
logger.info(f"AI Reply to {message.author}")
|
||||
prompt = message.clean_content
|
||||
if prompt == "" and message.embeds and message.embeds[0].description:
|
||||
prompt = message.embeds[0].description
|
||||
|
||||
answer = self.aibot.answer(prompt)
|
||||
if not isinstance(answer, str):
|
||||
logger.error(f"Got unexpected result from AIBot : {answer}")
|
||||
return
|
||||
|
||||
if len(answer) > 2000:
|
||||
logger.debug("Answer too long, sending as embed.")
|
||||
embed = discord.Embed(
|
||||
description=answer,
|
||||
thumbnail="https://mistral.ai/images/favicon/favicon-32x32.png",
|
||||
)
|
||||
await message.reply(embed=embed)
|
||||
else:
|
||||
await message.reply(answer)
|
||||
|
||||
async def react(self, message: discord.Message) -> None:
|
||||
emojis: tuple[str | discord.Emoji, ...] = self.uni_emojis
|
||||
if message.guild is not None and random.random() < 50 / 100:
|
||||
emojis = message.guild.emojis
|
||||
|
||||
emo: str | discord.Emoji = random.choice(emojis)
|
||||
await message.add_reaction(emo)
|
||||
|
||||
async def send_as_webhook(
|
||||
self,
|
||||
channel: discord.TextChannel,
|
||||
name: str,
|
||||
avatar_url: str | None,
|
||||
content: str,
|
||||
) -> None:
|
||||
webhooks = await channel.webhooks()
|
||||
webhook = discord.utils.get(webhooks, name="BotbotbotHook")
|
||||
|
||||
if webhook is None:
|
||||
webhook = await channel.create_webhook(name="BotbotbotHook")
|
||||
|
||||
await webhook.send(content=content, username=name, avatar_url=avatar_url)
|
||||
Loading…
Add table
Add a link
Reference in a new issue