Revert "Refactor animal image generation commands"

This reverts commit 866d496.

Author: NMertsch

prod-config.toml

@@ -67,3 +67,35 @@ required_role = "Organisers"
 [animals]
 channel_name = "animal-appreciation"
 cooldown_seconds = 10
+
+[animals.dog]
+error_messages = [
+    "The dogs are on strike today! Try again later. 🐾🪧",
+    "A wild error appeared! The dog got away... 🐕💨",
+    "Dog API is fetching a stick. Throw it again! 🦴",
+    "404: Dog not found. Have you checked under the couch? 🛋️",
+]
+
+[animals.cat]
+error_messages = [
+    "The cats are on strike today! Try again later. 🐾🪧",
+    "A wild error appeared! The cat got away... 😿💨",
+    "Cat API is fetching a mouse. Throw it again! 🦴",
+    "404: Cat not found. Have you checked inside a cardboard box? 📦",
+]
+
+[animals.duck]
+error_messages = [
+    "The ducks are on strike today! Try again later. 🐾🪧",
+    "A wild error appeared! The duck got away... 🦆💨",
+    "Duck API is in a bit of a pond situation. Try again later! 🦆",
+    "Quack! The duck API seems to be having an off day.",
+]
+
+[animals.fox]
+error_messages = [
+    "The foxes are on strike today! Try again later. 🐾🪧",
+    "A wild error appeared! The fox got away... 🦊💨",
+    "Fox API is fetching a chicken. Try again later! 🦴",
+    "404: Fox not found. Have you checked under the bush? 🌳",
+]

src/europython_discord/animals/clients.py

@@ -0,0 +1,111 @@
+from __future__ import annotations
+
+import logging
+import time
+from dataclasses import dataclass
+
+import aiohttp
+
+_logger = logging.getLogger(__name__)
+
+
+@dataclass
+class ImageResult:
+    url: str
+    source: str
+
+
+# APIs to fetch animals
+DOG_API_URL = "https://dog.ceo/api/breeds/image/random"
+DOG_SOURCE = "https://dog.ceo"
+CAT_API_URL = "https://cataas.com/cat/says/I%20love%20EuroPython"
+CAT_SOURCE = "https://cataas.com"
+DUCK_API_URL = "https://random-d.uk/api/randomimg"
+DUCK_SOURCE = "https://random-d.uk"
+FOX_API_URL = "https://randomfox.ca/floof/"
+FOX_SOURCE = "https://randomfox.ca/floof/"
+ANIMALITY_API_URL = "https://api.animality.xyz/img"
+ANIMALITY_SOURCE = "https://animality.xyz"
+
+
+class AnimalClient:
+    def __init__(self) -> None:
+        self._session = aiohttp.ClientSession()
+
+    async def fetch_image(self, animal: str) -> ImageResult | None:
+        """Fetch a random image for the given animal."""
+        if animal == "dog":
+            return await self._fetch_dog() or await self._fetch_animality(animal)
+        if animal == "cat":
+            return await self._fetch_cat() or await self._fetch_animality(animal)
+        if animal == "duck":
+            return await self._fetch_duck() or await self._fetch_animality(animal)
+        if animal == "fox":
+            return await self._fetch_fox() or await self._fetch_animality(animal)
+
+        return await self._fetch_animality(animal)
+
+    async def _fetch_dog(self) -> ImageResult | None:
+        try:
+            async with self._session.get(DOG_API_URL) as response:
+                response.raise_for_status()
+                data = await response.json()
+        except Exception:
+            _logger.exception("Failed to fetch dog image")
+            return None
+
+        url = data.get("message")
+        return ImageResult(url, DOG_SOURCE) if url else None
+
+    async def _fetch_fox(self) -> ImageResult | None:
+        try:
+            async with self._session.get(FOX_API_URL) as response:
+                response.raise_for_status()
+                data = await response.json()
+        except Exception:
+            _logger.exception("Failed to fetch fox image")
+            return None
+
+        url = data.get("image")
+        return ImageResult(url, FOX_SOURCE) if url else None
+
+    async def _fetch_duck(self) -> ImageResult | None:
+        timestamp = int(time.time() * 1000)
+        return ImageResult(f"{DUCK_API_URL}?t={timestamp}", DUCK_SOURCE)
+
+    async def _fetch_cat(self) -> ImageResult | None:
+        params = {
+            "position": "center",
+            "font": "Impact",
+            "fontSize": "50",
+            "fontColor": "#fff",
+            "fontBackground": "none",
+            "json": "true",
+        }
+        try:
+            async with self._session.get(CAT_API_URL, params=params) as response:
+                response.raise_for_status()
+                data = await response.json()
+        except Exception:
+            _logger.exception("Failed to fetch cat image")
+            return None
+
+        url = data.get("url")
+        if url:
+            if url.startswith("http"):
+                return ImageResult(url, CAT_SOURCE)
+            return ImageResult(f"https://cataas.com{url}", CAT_SOURCE)
+
+        return None
+
+    async def _fetch_animality(self, animal: str) -> ImageResult | None:
+        try:
+            async with self._session.get(f"{ANIMALITY_API_URL}/{animal}") as response:
+                response.raise_for_status()
+                data = await response.json()
+        except Exception:
+            _logger.exception("Failed to fetch animality image for %s", animal)
+            return None
+
+        url = data.get("image")
+        return ImageResult(url, ANIMALITY_SOURCE) if url else None

src/europython_discord/animals/cog.py

@@ -2,76 +2,103 @@
 
 import logging
 import random
-from typing import Self
+import time
+from collections import OrderedDict
 
 import discord
 from discord.ext import commands
 
-from europython_discord.animals import providers
-from europython_discord.animals.config import AnimalsConfig
-from europython_discord.animals.providers import AnimalImageProvider
-from europython_discord.animals.rate_limiter import RateLimiter
+from europython_discord.animals.clients import AnimalClient
+from europython_discord.animals.config import AnimalKind, AnimalsConfig
 
 _logger = logging.getLogger(__name__)
 
+_MAX_COOLDOWN_TRACKING = 100
+
 
 class AnimalsCog(commands.Cog):
     def __init__(
         self,
         bot: commands.Bot,
-        providers_by_animal: dict[str, list[AnimalImageProvider]],
-        rate_limiter: RateLimiter,
-        channel_name: str,
+        config: AnimalsConfig,
+        client: AnimalClient | None = None,
     ) -> None:
-        self._bot = bot
-        self._channel_name = channel_name
-        self._providers = providers_by_animal
-        self._rate_limiter = rate_limiter
-
-        for animal in self._providers:
-            self._bot.add_command(self._create_animal_command(animal))
-
+        self.bot = bot
+        self.config = config
+        self._client = client or AnimalClient()
+        self._last_usage_timestamp_by_user_id: OrderedDict[int, float] = OrderedDict()
         _logger.info("Cog 'Animals' has been initialized")
 
-    async def post_animal_picture(self, animal: str, ctx: commands.Context) -> None:
-        if ctx.channel.name != self._channel_name:
-            return
-        if self._rate_limiter.is_rate_limited(ctx.author.id):
-            return
-        if animal not in self._providers:
+    async def handle_animal_command(self, ctx: commands.Context, animal: AnimalKind) -> None:
+        if ctx.channel.name != self.config.channel_name:
             return
 
-        provider = random.choice(self._providers[animal])  # noqa: S311 suspicious-non-cryptographic-random-usage
-        try:
-            image = await provider.generate_image()
-        except Exception:
-            _logger.exception("Error while generating %s image", animal)
-            await ctx.send(f"Failed to fetch {animal} picture. Internal error, please report it.")
+        if self._is_rate_limited(ctx.author.id):
             return
 
-        if image is None:
-            _logger.error("Failed to fetch %s pictures", animal)
-            await ctx.send(f"Failed to fetch {animal} picture.")
+        result = await self._client.fetch_image(animal)
+        if result is not None:
+            embed = discord.Embed()
+            embed.description = f"Behold! A friendly {animal} appeared from {result.source}"
+            embed.set_image(url=result.url)
+            self._update_rate_limit_cache(ctx.author.id)
+            await ctx.send(embed=embed)
             return
 
-        embed = discord.Embed()
-        embed.description = f"Behold! A friendly {animal} appeared from {image.source}"
-        embed.set_image(url=image.url)
-        self._rate_limiter.register_usage(ctx.author.id)
-        await ctx.send(embed=embed)
-
-    def _create_animal_command(self, animal: str) -> commands.HybridCommand:
-        @commands.hybrid_command(name=animal, description=f"Get a random {animal} picture")
-        async def callback(ctx: commands.Context) -> None:
-            await self.post_animal_picture(animal, ctx)
-
-        return callback
-
-    @classmethod
-    def from_config(cls, bot: commands.Bot, config: AnimalsConfig) -> Self:
-        return cls(
-            bot=bot,
-            providers_by_animal=providers.get_all_providers(),
-            rate_limiter=RateLimiter(config.cooldown_seconds),
-            channel_name=config.channel_name,
+        await ctx.send(self._error_message_for(animal))
+
+    @commands.hybrid_command(name="dog", description="Get a random dog picture")
+    async def dog_command(self, ctx: commands.Context) -> None:
+        await self.handle_animal_command(ctx, AnimalKind.DOG)
+
+    @commands.hybrid_command(name="cat", description="Get a random cat picture")
+    async def cat_command(self, ctx: commands.Context) -> None:
+        await self.handle_animal_command(ctx, AnimalKind.CAT)
+
+    @commands.hybrid_command(name="duck", description="Get a random duck picture")
+    async def duck_command(self, ctx: commands.Context) -> None:
+        await self.handle_animal_command(ctx, AnimalKind.DUCK)
+
+    @commands.hybrid_command(name="fox", description="Get a random fox picture")
+    async def fox_command(self, ctx: commands.Context) -> None:
+        await self.handle_animal_command(ctx, AnimalKind.FOX)
+
+    async def cog_load(self) -> None:
+        existing = {"dog", "cat", "duck", "fox"}
+        for animal in AnimalKind:
+            if animal in existing:
+                continue
+            cmd = _make_animality_command(animal, self)
+            self.bot.add_command(cmd)
+
+    def _error_message_for(self, animal: AnimalKind) -> str:
+        if animal_config := self.config.config_by_kind.get(animal):
+            return random.choice(animal_config.error_messages)  # noqa: S311
+
+        if not self.config.animality_error_messages:
+            return f"Couldn't find a {animal} picture right now."
+        plural = f"{animal}s" if animal != "fish" else "fish"
+        return random.choice(self.config.animality_error_messages).format(  # noqa: S311
+            animal=animal, plural=plural
         )
+
+    def _is_rate_limited(self, user_id: int) -> bool:
+        last_usage_timestamp = self._last_usage_timestamp_by_user_id.get(user_id, 0)
+        return last_usage_timestamp + self.config.cooldown_seconds > time.time()
+
+    def _update_rate_limit_cache(self, user_id: int) -> None:
+        # update cache
+        self._last_usage_timestamp_by_user_id[user_id] = time.time()
+
+        # trim cache
+        self._last_usage_timestamp_by_user_id.move_to_end(user_id)
+        if len(self._last_usage_timestamp_by_user_id) > _MAX_COOLDOWN_TRACKING:
+            self._last_usage_timestamp_by_user_id.popitem(last=False)
+
+
+def _make_animality_command(name: AnimalKind, cog: AnimalsCog) -> commands.HybridCommand:
+    @commands.hybrid_command(name=name, description=f"Get a random {name} picture")
+    async def callback(ctx: commands.Context) -> None:
+        await cog.handle_animal_command(ctx, name)
+
+    return callback

src/europython_discord/animals/config.py

@@ -1,8 +1,60 @@
 from __future__ import annotations
 
+from enum import StrEnum
+
 from pydantic import BaseModel
 
 
+class AnimalKind(StrEnum):
+    CAT = "cat"
+    DOG = "dog"
+    BIRD = "bird"
+    PANDA = "panda"
+    REDPANDA = "redpanda"
+    KOALA = "koala"
+    FOX = "fox"
+    WHALE = "whale"
+    DOLPHIN = "dolphin"
+    KANGAROO = "kangaroo"
+    RABBIT = "rabbit"
+    LION = "lion"
+    BEAR = "bear"
+    FROG = "frog"
+    DUCK = "duck"
+    PENGUIN = "penguin"
+    AXOLOTL = "axolotl"
+    CAPYBARA = "capybara"
+    HEDGEHOG = "hedgehog"
+    TURTLE = "turtle"
+    NARWHAL = "narwhal"
+    SQUIRREL = "squirrel"
+    FISH = "fish"
+    HORSE = "horse"
+
+
+class AnimalSpecificConfig(BaseModel):
+    error_messages: list[str]
+
+
 class AnimalsConfig(BaseModel):
     channel_name: str
     cooldown_seconds: int
+    dog: AnimalSpecificConfig
+    cat: AnimalSpecificConfig
+    duck: AnimalSpecificConfig
+    fox: AnimalSpecificConfig
+    animality_error_messages: list[str] = [
+        "The {plural} are on strike today! Try again later.",
+        "A wild error appeared! The {animal} got away...",
+        "The {animal} API is fetching a treat. Try again!",
+        "404: {plural} not found. Have you checked the nearest zoo?",
+    ]
+
+    @property
+    def config_by_kind(self) -> dict[AnimalKind, AnimalSpecificConfig]:
+        return {
+            AnimalKind.DOG: self.dog,
+            AnimalKind.CAT: self.cat,
+            AnimalKind.DUCK: self.duck,
+            AnimalKind.FOX: self.fox,
+        }