Improved caching and utils structure

modules/utils/__init__.py

@@ -0,0 +1,8 @@
+from .autocomplete_utils import (
+    autocomplete_active_events,
+    autocomplete_event_stages,
+    autocomplete_languages,
+    autocomplete_timezones,
+)
+from .cache_utils import restore_from_cache
+from .logging_utils import get_logger, get_logging_config

modules/utils/autocomplete_utils.py

@@ -0,0 +1,62 @@
+from datetime import datetime
+from typing import Any, Dict, List
+from zoneinfo import ZoneInfo, available_timezones
+
+from bson import ObjectId
+from discord import AutocompleteContext, OptionChoice
+from pymongo import ASCENDING
+
+from modules.database import col_events, col_stages
+
+
+async def autocomplete_timezones(ctx: AutocompleteContext) -> List[str]:
+    """Return available timezones"""
+
+    return sorted(list(available_timezones()))
+
+
+async def autocomplete_languages(ctx: AutocompleteContext) -> List[str]:
+    """Return locales supported by the bot"""
+
+    # TODO Discord normally uses a different set of locales.
+    #  For example, "en" being "en-US", etc. This will require changes to locale handling later.
+    return ctx.bot.locales.keys()
+
+
+async def autocomplete_active_events(ctx: AutocompleteContext) -> List[OptionChoice]:
+    """Return list of active events"""
+
+    query: Dict[str, Any] = {
+        "ended": None,
+        "ends": {"$gt": datetime.now(tz=ZoneInfo("UTC"))},
+        "cancelled": {"$ne": True},
+    }
+
+    event_names: List[OptionChoice] = []
+
+    async for result in col_events.find(query):
+        event_names.append(OptionChoice(result["name"], str(result["_id"])))
+
+    return event_names
+
+
+async def autocomplete_event_stages(ctx: AutocompleteContext) -> List[OptionChoice]:
+    """Return list of stages of the event"""
+
+    event_id: str | None = ctx.options["event"]
+
+    if event_id is None:
+        return []
+
+    query: Dict[str, Any] = {
+        "event_id": ObjectId(event_id),
+    }
+
+    event_stages: List[OptionChoice] = []
+
+    async for result in col_stages.find(query).sort([("sequence", ASCENDING)]):
+        event_stages.append(
+            OptionChoice(f"{result['sequence']+1} ({result['question']})", str(result["_id"]))
+        )
+
+    return event_stages

modules/utils/cache_utils.py

@@ -0,0 +1,10 @@
+from typing import Any, Dict, Optional
+
+from bson import ObjectId
+from libbot.cache.classes import Cache
+
+
+def restore_from_cache(
+    cache_prefix: str, cache_key: str | int | ObjectId, cache: Optional[Cache] = None
+) -> Dict[str, Any] | None:
+    return None if cache is None else cache.get_json(f"{cache_prefix}_{cache_key}")

modules/utils/color.py

@@ -0,0 +1,6 @@
+def hex_to_int(hex_color: str) -> int:
+    return int(hex_color.lstrip("#"), 16)
+
+
+def int_to_hex(integer_color: int) -> str:
+    return "#" + format(integer_color, "06x")

modules/utils/logging_utils.py

@@ -0,0 +1,35 @@
+import logging
+from logging import Logger
+from pathlib import Path
+from typing import Any, Dict
+
+from libbot.utils import config_get
+
+
+def get_logging_config() -> Dict[str, Any]:
+    return {
+        "version": 1,
+        "disable_existing_loggers": False,
+        "handlers": {
+            "file": {
+                "class": "logging.handlers.RotatingFileHandler",
+                "filename": str(Path("logs/latest.log")),
+                "maxBytes": 500000,
+                "backupCount": 10,
+                "formatter": "simple",
+            },
+            "console": {"class": "logging.StreamHandler", "formatter": "systemd"},
+        },
+        "formatters": {
+            "simple": {"format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s"},
+            "systemd": {"format": "%(name)s - %(levelname)s - %(message)s"},
+        },
+        "root": {
+            "level": "DEBUG" if config_get("debug") else "INFO",
+            "handlers": ["file", "console"],
+        },
+    }
+
+
+def get_logger(name: str) -> Logger:
+    return logging.getLogger(name)