feat(cogs): ✨ Add translate module

This module automatically translates foreign-language chat messages to the configured target language (English by default) using the AGPL-licensed, self-hostable LibreTranslate.

kolkra_ng/bot.py

@@ -29,7 +29,8 @@ class Kolkra(commands.Bot):
     def __init__(self, config: Config) -> None:
         super().__init__(
             command_prefix=commands.when_mentioned_or(";"),
-            intents=Intents.default() | Intents(members=True, message_content=True),
+            intents=Intents.default()
+            | Intents(members=True, message_content=True, presences=True),
         )
         self.config = config
         self.motor = AsyncIOMotorClient(
@@ -83,6 +84,8 @@ async def load_modules(self) -> None:
             )
             try:
                 await self.load_extension(module)
+            except NotImplementedError:
+                log.info("Skipped unfinished module %s", module)
             except Exception as e:
                 log.warning("Failed to load module %s", module, exc_info=e)
             else:

kolkra_ng/cogs/mod/mod_actions/warning.py

@@ -66,7 +66,9 @@ async def dm_embed(self, bot: Kolkra) -> Embed:
         return embed
 
     def log_base(self) -> Embed:
-        return Embed(title="Warning", color=Color.yellow()).set_thumbnail(url=icons8("error"))
+        return Embed(title="Warning", color=Color.yellow()).set_thumbnail(
+            url=icons8("error")
+        )
 
     async def log_embed(self) -> Embed:
         count = await self.cached_count()

kolkra_ng/cogs/translate/__init__.py

@@ -0,0 +1,128 @@
+"""Because every other bot we've tried can go f**k themselves.
+"""
+
+import logging
+from uuid import UUID
+
+from aiohttp import ClientSession
+from discord import Color, Embed, Message
+from discord.ext import commands
+from pydantic import BaseModel, Field, HttpUrl, Secret
+from pydantic_extra_types.language_code import LanguageAlpha2
+
+from kolkra_ng.bot import Kolkra
+from kolkra_ng.cogs.translate.api import (
+    DetectRequest,
+    DetectResponseItem,
+    LanguagesResponseItem,
+    TranslateRequest,
+    TranslateResponse,
+)
+from kolkra_ng.embeds import WarningEmbed, icons8
+
+log = logging.getLogger(__name__)
+
+
+class TranslateConfig(BaseModel):
+    target_language: LanguageAlpha2 = Field(
+        default="en",
+        description="The language to translate foreign-language messages to. Defaults to 'en' (English).",
+    )
+    api_base_url: HttpUrl = Field(
+        description="Base URL of a [LibreTranslate](https://github.com/LibreTranslate/LibreTranslate) instance.",
+    )
+    api_key: Secret[UUID] | None = Field(
+        default=None,
+        description="An API key for the LibreTranslate instance, if required.",
+    )
+    aiohttp_params: dict = Field(
+        default_factory=dict,
+        description="Extra parameters to pass to the aiohttp.ClientSession constructor.",
+    )
+
+
+class TranslateCog(commands.Cog):
+    def __init__(self, bot: Kolkra) -> None:
+        super().__init__()
+        self.bot = bot
+        self.config = TranslateConfig(**bot.config.cogs.get(self.__cog_name__, {}))
+        self.session = ClientSession(
+            base_url=self.config.api_base_url.unicode_string(),
+            **self.config.aiohttp_params,
+        )
+
+    async def cog_load(self) -> None:
+        async with self.session.get("/languages") as resp:
+            resp.raise_for_status()
+            languages_response = [
+                LanguagesResponseItem(**item) for item in await resp.json()
+            ]
+        self.all_languages = {
+            language.code: language for language in languages_response
+        }
+        self.supported_languages = {
+            code: language
+            for code, language in self.all_languages.items()
+            if self.config.target_language in language.targets
+        }
+
+    async def cog_unload(self) -> None:
+        await self.session.close()
+
+    @commands.Cog.listener()
+    async def on_message(self, message: Message) -> None:
+        if message.author.bot or not message.content:
+            return
+        async with self.session.post(
+            "/detect",
+            data=DetectRequest(
+                q=message.content,
+                api_key=(
+                    key.get_secret_value() if (key := self.config.api_key) else None
+                ),
+            ).model_dump(mode="json", exclude_none=True),
+        ) as resp:
+            resp.raise_for_status()
+            detect_response = [DetectResponseItem(**item) for item in await resp.json()]
+        detected_language = max(detect_response, key=lambda x: x.confidence)
+        if detected_language.language == self.config.target_language:
+            return
+        elif detected_language.language not in self.supported_languages:
+            await message.reply(
+                embed=WarningEmbed(
+                    title="Language not supported!",
+                    description=f"This message's detected language ({self.all_languages[detected_language.language].name}) cannot be translated to {self.all_languages[detected_language.language].name}.",
+                ),
+                mention_author=False,
+            )
+            return
+        async with self.session.post(
+            "/translate",
+            data=TranslateRequest(
+                q=message.content,
+                api_key=(
+                    key.get_secret_value() if (key := self.config.api_key) else None
+                ),
+                source=detected_language.language,
+                target=self.config.target_language,
+            ).model_dump(mode="json", exclude_none=True),
+        ) as resp:
+            resp.raise_for_status()
+            translate_response = TranslateResponse(**await resp.json())
+        await message.reply(
+            embed=Embed(
+                color=Color.blue(),
+                title="Automatic translation",
+                description=translate_response.translatedText,
+            )
+            .add_field(
+                name="Language",
+                value=f"{self.all_languages[detected_language.language].name} -> {self.all_languages[self.config.target_language].name}",
+            )
+            .set_thumbnail(url=icons8("translate-text"))
+            .set_footer(text="Machine translations may not be 100% accurate.")
+        )
+
+
+async def setup(bot: Kolkra) -> None:
+    await bot.add_cog(TranslateCog(bot))

kolkra_ng/cogs/translate/api.py

@@ -0,0 +1,36 @@
+from typing import Literal
+from uuid import UUID
+
+from pydantic import BaseModel, Field, NonNegativeInt
+from pydantic_extra_types.language_code import LanguageAlpha2
+
+
+class DetectRequest(BaseModel):
+    q: str
+    api_key: UUID | None
+
+
+class DetectResponseItem(BaseModel):
+    confidence: int = Field(ge=0, le=100)
+    language: LanguageAlpha2
+
+
+class LanguagesResponseItem(BaseModel):
+    code: LanguageAlpha2
+    name: str
+    targets: set[LanguageAlpha2]
+
+
+class TranslateRequest(BaseModel):
+    q: str
+    source: LanguageAlpha2 | Literal["auto"]
+    target: LanguageAlpha2
+    format: Literal["text", "html"] = "text"
+    alternatives: NonNegativeInt = 0
+    api_key: UUID | None
+
+
+class TranslateResponse(BaseModel):
+    translatedText: str
+    detectedLanguage: DetectResponseItem | None = None
+    alternatives: list[str] | None = None

kolkra_ng/enums/by_name.py

@@ -36,6 +36,7 @@ def __get_pydantic_json_schema__(
 
 
 def by_name(cls: E) -> E:
+    """Decorator to make Pydantic handle an enum by name rather than by value."""
 
     cls.__get_pydantic_core_schema__ = (  # pyright: ignore # noqa: PGH003
         types.MethodType(__get_pydantic_core_schema__, cls)