Reworked AI: cog, initiative system w/ task managment

Author: Ambratolm

bot/cogs/ai_cog.py

@@ -1,10 +1,7 @@
-import asyncio
-
 from discord import Color, Embed, Guild, Interaction, Member, User, app_commands
 from discord.ext.commands import Cog
-from humanize import intcomma, intword, metric, naturalsize, naturaltime, ordinal
+from humanize import intcomma, naturalsize, naturaltime
 from odmantic import query
-from tabulate import tabulate
 
 from bot.main import ActBot
 from bot.ui import EmbedX

bot/cogs/board_cog.py

@@ -1,9 +1,8 @@
-from discord import Interaction, app_commands
+from discord import ClientException, Interaction, VoiceChannel, app_commands
 from discord.ext.commands import Cog
 
 from bot.main import ActBot
 from bot.ui import EmbedX
-from db.actor import Actor
 
 
 # ----------------------------------------------------------------------------------------------------
@@ -30,21 +29,24 @@ async def sync(self, interaction: Interaction, global_sync: bool = True):
             ephemeral=True,
         )
 
-    # @app_commands.guild_only()
-    # @app_commands.checks.has_permissions(administrator=True)
-    # @app_commands.command(description="Synchronize commands")
-    # async def update_actors(self, interaction: Interaction):
-    #     if not interaction.guild:
-    #         return
-    #     await interaction.response.defer(ephemeral=True)
-    #     db = self.bot.get_db(interaction.guild)
-    #     actors = db.find(Actor)
-    #     members = interaction.guild.members
-    #     for member in members:
-    #         for actor in actors:
-    #             if actor.id == member.id:
-
-    #         if not actor:
-    #             actor = self.bot.create_actor(member)
-    #     db.save_all(actors)
-    #     await interaction.followup.send(embed=EmbedX.success(title="", description=f"{len(members)}/{len(actors)} actors updated."))
+    # ----------------------------------------------------------------------------------------------------
+    # * Join
+    # ----------------------------------------------------------------------------------------------------
+    @app_commands.default_permissions(administrator=True)
+    @app_commands.checks.has_permissions(administrator=True)
+    @app_commands.command(description="Add bot to a voice channel")
+    async def join(self, interaction: Interaction, channel: VoiceChannel):
+        try:
+            await channel.connect()
+            await interaction.response.send_message(
+                embed=EmbedX.info(f"Joined {channel.name}."), ephemeral=True
+            )
+        except ClientException:
+            await interaction.response.send_message(
+                embed=EmbedX.warning("Already in a voice channel."), ephemeral=True
+            )
+        except Exception as e:
+            await interaction.response.send_message(
+                embed=EmbedX.error(f"Could not join the voice channel: {e}"),
+                ephemeral=True,
+            )

bot/cogs/console_cog.py

@@ -1,7 +1,7 @@
 import re
 from random import randint
 
-from discord import Color, Embed, Member, Message, Role, utils
+from discord import Member, Message, Role, utils
 from discord.ext.commands import Cog
 
 from bot.main import ActBot

bot/cogs/farm_cog.py

@@ -1,7 +1,7 @@
 from collections import defaultdict
 from datetime import timedelta
 
-from discord import Color, Embed, Member, Message
+from discord import Member, Message
 from discord.ext.commands import Cog
 from profanity_check import predict_prob
 
@@ -17,7 +17,7 @@
 # * Filter Cog
 # ----------------------------------------------------------------------------------------------------
 class FilterCog(Cog, description="Filters blacklisted message content."):
-    TOLERANCE = 0.99  # Between 0 and 1
+    TOLERANCE = 0.99  # 99 %
     MAX_OFFENSES = 5
     GOLD_PENALTY = 500
 
@@ -100,9 +100,11 @@ async def on_message(self, message: Message):
     # ----------------------------------------------------------------------------------------------------
 
     @classmethod
-    def get_profane_words(cls, words: list[str]) -> list[str]:
+    def get_profane_words(cls, words: list[str]) -> list[str] | None:
         """Get list of profane words from given list. If non found, get empty list."""
-        predictions = predict_prob(words)
+        predictions = predict_prob(words) if words else None
+        if predictions is None:
+            return None
         profane_words = []
         for i, word in enumerate(words):
             if predictions[i] >= cls.TOLERANCE:  # 1 means profane, 0 means clean

bot/cogs/filter_cog.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from typing import ClassVar, Optional
+from typing import Any, ClassVar, Optional
 
 from odmantic import Field, Model
 from pydantic import NonNegativeInt
@@ -19,6 +19,7 @@ class Actor(Model):
     avatar_url: str = ""
 
     ai_interacted_at: Optional[datetime] = None  # Last time actor interacted with AI
+    ai_chat_history: list[dict[str, Any]] = []
 
     xp: NonNegativeInt = 0
     level: NonNegativeInt = 0

bot/cogs/purge_cog.py

@@ -14,7 +14,7 @@
 
 
 # ----------------------------------------------------------------------------------------------------
-# * Database reference
+# * Database Reference
 # ----------------------------------------------------------------------------------------------------
 class DbRef(Model):
     """Database model for storage of reference and relevant data about a database."""

db/actor.py

@@ -14,6 +14,7 @@ dependencies = [
     "humanize>=4.11.0",
     "odmantic>=1.0.2",
     "pydantic>=2.10.6",
+    "pynacl>=1.5.0",
     "python-dotenv>=1.0.1",
     "tabulate>=0.9.0",
 ]

db/main.py

@@ -67,7 +67,7 @@ def use_session(self, id: int, history: list[Content] | None = None) -> Chat:
                 ]  # Sanitize history to ensure correct dumping in dump_history()
             chat = self._client.chats.create(model=self.model_name, history=history)  # type: ignore
             self._chats[id] = chat
-            self._current_chat_id = id
+        self._current_chat_id = id
         return chat
 
     def dump_history(self, id: int | None = None, history_max_items=20) -> list[dict]: