API: utils

utils.EmojiManager

EmojiManager

Loads and resolves emoji placeholders for global and module scopes.

Placeholders follow the pattern :emoji_name:. This manager supports a global emoji catalog (single JSON file) and per‑module catalogs.

Source code in utils/EmojiManager.py

class EmojiManager:
    """Loads and resolves emoji placeholders for global and module scopes.

    Placeholders follow the pattern `:emoji_name:`. This manager supports a
    global emoji catalog (single JSON file) and per‑module catalogs.
    """

    EMOJI_PATTERN = re.compile(r":([a-zA-Z0-9_]+):")

    def __init__(self, bot, global_path: str, logger: Logger):
        """Initialize the emoji manager.

        Args:
            bot: The bot instance (for context; not required to resolve emojis).
            global_path: Directory containing the `emojis.json` file.
            logger: Logger for diagnostics.
        """
        self.bot = bot
        self.logger = logger
        self.global_path = Path(global_path)
        self.global_emojis: Dict[str, str] = {}  # Emojis globais
        self.module_emojis: Dict[str, Dict[str, str]] = {}  # Emojis por módulo

    def load_global_emojis(self):
        """Load global emojis from `<global_path>/emojis.json`."""
        path = self.global_path / "emojis.json"
        if path.exists():
            with open(path, "r", encoding="utf-8") as f:
                try:
                    self.global_emojis = json.load(f)
                    self.logger.info("Global emojis loaded with success.")
                except json.JSONDecodeError as e:
                    self.logger.error(f"Erro while loading global emojis: {e}")
        else:
            self.logger.warning("File emojis.json wasn't find for global emojis.")

    def load_module_emojis(self, module_name: str, module_path: str):
        """Load per‑module emojis from `<module_path>/emojis.json`.

        Args:
            module_name: Name of the module to associate the catalog.
            module_path: Filesystem path to the module root.
        """
        path = Path(module_path) / "emojis.json"
        if path.exists():
            with open(path, "r", encoding="utf-8") as f:
                try:
                    self.module_emojis[module_name] = json.load(f)
                    self.logger.info(f"Emojis loaded with succes for the module: {module_name}")
                except json.JSONDecodeError as e:
                    self.logger.error(f"Erro while loading emojis for the module {module_name}: {e}")
        else:
            self.logger.warning(f"File emojis.json not found for the module: {module_name}")

    def replace_emojis(self, text: str, module_name: Optional[str] = None) -> str:
        """Replace `:emoji_name:` placeholders with actual emojis.

        Args:
            text: The source text that may contain placeholders.
            module_name: Optional module name to use a module‑specific catalog.

        Returns:
            The processed text with placeholders substituted.
        """

        def emoji_replacer(match):
            emoji_name = match.group(1)
            if module_name and module_name in self.module_emojis:
                return self.module_emojis[module_name].get(emoji_name, f":{emoji_name}:")
            return self.global_emojis.get(emoji_name, f":{emoji_name}:")

        return self.EMOJI_PATTERN.sub(emoji_replacer, text)

__init__(bot, global_path, logger)

Initialize the emoji manager.

Parameters:

Name	Type	Description	Default
bot		The bot instance (for context; not required to resolve emojis).	required
global_path	str	Directory containing the emojis.json file.	required
logger	Logger	Logger for diagnostics.	required

Source code in utils/EmojiManager.py

def __init__(self, bot, global_path: str, logger: Logger):
    """Initialize the emoji manager.

    Args:
        bot: The bot instance (for context; not required to resolve emojis).
        global_path: Directory containing the `emojis.json` file.
        logger: Logger for diagnostics.
    """
    self.bot = bot
    self.logger = logger
    self.global_path = Path(global_path)
    self.global_emojis: Dict[str, str] = {}  # Emojis globais
    self.module_emojis: Dict[str, Dict[str, str]] = {}  # Emojis por módulo

load_global_emojis()

Load global emojis from <global_path>/emojis.json.

Source code in utils/EmojiManager.py

def load_global_emojis(self):
    """Load global emojis from `<global_path>/emojis.json`."""
    path = self.global_path / "emojis.json"
    if path.exists():
        with open(path, "r", encoding="utf-8") as f:
            try:
                self.global_emojis = json.load(f)
                self.logger.info("Global emojis loaded with success.")
            except json.JSONDecodeError as e:
                self.logger.error(f"Erro while loading global emojis: {e}")
    else:
        self.logger.warning("File emojis.json wasn't find for global emojis.")

load_module_emojis(module_name, module_path)

Load per‑module emojis from <module_path>/emojis.json.

Parameters:

Name	Type	Description	Default
module_name	str	Name of the module to associate the catalog.	required
module_path	str	Filesystem path to the module root.	required

Source code in utils/EmojiManager.py

def load_module_emojis(self, module_name: str, module_path: str):
    """Load per‑module emojis from `<module_path>/emojis.json`.

    Args:
        module_name: Name of the module to associate the catalog.
        module_path: Filesystem path to the module root.
    """
    path = Path(module_path) / "emojis.json"
    if path.exists():
        with open(path, "r", encoding="utf-8") as f:
            try:
                self.module_emojis[module_name] = json.load(f)
                self.logger.info(f"Emojis loaded with succes for the module: {module_name}")
            except json.JSONDecodeError as e:
                self.logger.error(f"Erro while loading emojis for the module {module_name}: {e}")
    else:
        self.logger.warning(f"File emojis.json not found for the module: {module_name}")

replace_emojis(text, module_name=None)

Replace :emoji_name: placeholders with actual emojis.

Parameters:

Name	Type	Description	Default
text	str	The source text that may contain placeholders.	required
module_name	Optional[str]	Optional module name to use a module‑specific catalog.	None

Returns:

Type	Description
str	The processed text with placeholders substituted.

Source code in utils/EmojiManager.py

def replace_emojis(self, text: str, module_name: Optional[str] = None) -> str:
    """Replace `:emoji_name:` placeholders with actual emojis.

    Args:
        text: The source text that may contain placeholders.
        module_name: Optional module name to use a module‑specific catalog.

    Returns:
        The processed text with placeholders substituted.
    """

    def emoji_replacer(match):
        emoji_name = match.group(1)
        if module_name and module_name in self.module_emojis:
            return self.module_emojis[module_name].get(emoji_name, f":{emoji_name}:")
        return self.global_emojis.get(emoji_name, f":{emoji_name}:")

    return self.EMOJI_PATTERN.sub(emoji_replacer, text)

utils.Translator

Translator

High‑level translation service with caching and emoji post‑processing.

This utility

• Resolves the preferred language per guild (with caching).
• Loads and caches global translation files from a shared folder.
• Loads and caches module translation files from each module folder.
• Replaces emoji placeholders using EmojiManager (if available).
• Exposes convenience helpers to retrieve a translation or a translator function (sync/async).

Notes

• Global translations live under global_path/*.json.
• Module translations are discovered under <module>/translations/*.json unless a different folder is declared in the module's manifest.

Source code in utils/Translator.py

class Translator:
    """High‑level translation service with caching and emoji post‑processing.

    This utility:
      * Resolves the preferred language per guild (with caching).
      * Loads and caches **global** translation files from a shared folder.
      * Loads and caches **module** translation files from each module folder.
      * Replaces emoji placeholders using `EmojiManager` (if available).
      * Exposes convenience helpers to retrieve a translation or a translator
        function (sync/async).

    Notes:
        * Global translations live under `global_path/*.json`.
        * Module translations are discovered under `<module>/translations/*.json`
          unless a different folder is declared in the module's manifest.
    """
    def __init__(self, bot: ExtendedClient, global_path: str, logger: logging.Logger):
        """Initialize the translator and its caches.

        Args:
            bot: The extended Discord client.
            global_path: Filesystem path to the global translations directory.
            logger: Logger for diagnostics.
        """
        self.bot = bot
        self.logger = logger
        self.global_path = Path(global_path)
        self.global_translations_cache = {}
        self.module_translation_cache = {}
        self.language_cache = {}

    async def get_language(self, guild_id: str) -> str:
        """Return the configured language for a guild (default `'en'`).

        Normalizes common aliases like `"English"`, `"en-US"`, `"pt-br"`, etc.
        Results are cached per guild id.

        Args:
            guild_id: Discord guild id.

        Returns:
            Normalized language code such as `"en"` or `"pt"`.
        """
        guild_id = str(guild_id)
        if guild_id in self.language_cache:
            return self.language_cache[guild_id]

        guild = await self.bot.guild_manager.fetch_or_create(guild_id)
        language = guild.data.get("language", "en")

        if language in ["Inglês", "English", "en", "en-US"]:
            self.language_cache[guild_id] = "en"
            language = "en"
        elif language in ["Português", "pt-br", "Português (Brasileiro)", "pt"]:
            self.language_cache[guild_id] = "pt"
            language = "pt"
        else:
            self.language_cache[guild_id] = language
        return language

    def get_language_sync(self, guild_id: Optional[str]) -> str:
        """Synchronous variant of `get_language` using the local cache only.

        If the language for `guild_id` is not cached, returns `'en'`.

        Args:
            guild_id: Guild id (or `None`).

        Returns:
            A normalized language code.
        """
        if guild_id and guild_id in self.language_cache:
            guild_id = str(guild_id)
            return self.language_cache[guild_id]
        return "en"

    def update_language_cache(self, guild_id: str, language: str):
        """Update the local language cache for a guild.

        Args:
            guild_id: Guild id.
            language: Normalized language code to store.
        """
        guild_id = str(guild_id)
        self.language_cache[guild_id] = language

    def _process_emojis(self, text: str, module_name: Optional[str] = None) -> str:
        """Replace emoji placeholders (`:name:`) in a string.

        Delegates to `EmojiManager.replace_emojis` if the bot exposes an
        `emoji_manager` attribute. Otherwise returns the text unchanged.

        Args:
            text: Source text that may contain placeholders.
            module_name: Module name for module‑scoped emoji lookups.

        Returns:
            Text with emoji placeholders resolved.
        """
        if hasattr(self.bot, "emoji_manager"):
            return self.bot.emoji_manager.replace_emojis(text, module_name)
        return text

    def _process_translation(self, translations: Dict[str, Any], module_name: Optional[str] = None) -> Dict[str, Any]:
        """Recursively apply emoji replacements to a translation tree.

        Args:
            translations: Nested mapping of translation keys → values.
            module_name: Module name for module‑scoped emoji lookups.

        Returns:
            A new mapping with emojis processed.
        """
        processed = {}
        for key, value in translations.items():
            if isinstance(value, str):
                processed[key] = self._process_emojis(value, module_name)
            elif isinstance(value, dict):
                processed[key] = self._process_translation(value, module_name)
            else:
                processed[key] = value
        return processed

    async def refresh_translation_cache(self):
        """Reload global and module translation caches asynchronously.

        Scans the `global_path` for `*.json` files and each loaded module's
        translations directory (from its manifest or default `"translations"`).
        """

        available_languages = {file.stem for file in self.global_path.glob("*.json")}
        for language in available_languages:
            path = self.global_path / f"{language}.json"
            if path.exists():
                try:
                    async with aiofiles.open(path, "r", encoding="utf-8") as f:
                        content = await f.read()
                        translations = json.loads(content)
                        self.global_translations_cache[language] = self._process_translation(translations)
                        self.logger.info(f"Global translations loaded for language: {language}")
                except json.JSONDecodeError as e:
                    self.logger.error(f"Error loading global translations for {language}: {e}")

        # Carregar traduções dos módulos
        for module_name, module in self.bot.modules.items():
            translations_folder = module.data.get("translationsFolder") or "translations"
            translations_path = Path(module.path) / translations_folder
            available_languages = {file.stem for file in translations_path.glob("*.json")}
            for language in available_languages:
                path = translations_path / f"{language}.json"
                if path.exists():
                    try:
                        async with aiofiles.open(path, "r", encoding="utf-8") as f:
                            content = await f.read()
                            translations = json.loads(content)
                            processed_translations = self._process_translation(translations, module_name)
                            cache_key = f"{module_name}:{language}"
                            self.module_translation_cache[cache_key] = processed_translations
                            self.logger.info(f"Module translations loaded for {module_name}, language: {language}")
                    except json.JSONDecodeError as e:
                        self.logger.error(f"Error loading module translations for {module_name}, language: {language}: {e}")

    def get_translation(self, key: str, language: str, module_name: Optional[str] = None) -> str:
        """Return the processed translation for a dot‑separated key.

        Args:
            key: Hierarchical key such as `"help.commands.ping.description"`.
            language: Target language code.
            module_name: Module name to search in module cache; if omitted,
                the global cache is used.

        Returns:
            The resolved translation string, or `key` if not found.
        """
        translations = (
            self.module_translation_cache.get(f"{module_name}:{language}", {})
            if module_name
            else self.global_translations_cache.get(language, {})
        )

        keys = key.split(".")
        for k in keys:
            if isinstance(translations, dict):
                translations = translations.get(k)
            else:
                return key  # Retorna a própria chave se não encontrada
        return translations if isinstance(translations, str) else key


    async def get_translator(self, guild_id: str, module_name: Optional[str] = None) -> Callable[[str], str]:
        """Return a callable that resolves translations for a guild asynchronously.

        The callable captures the guild's current language and interpolates `**kwargs`
        with standard Python `str.format`.

        Args:
            guild_id: Guild id to derive the language from.
            module_name: Module name for module‑scoped translations.

        Returns:
            A function `fn(key: str, **kwargs) -> str`.
        """
        guild_id = str(guild_id)
        language = await self.get_language(guild_id=guild_id)

        def translator_func(key: str, **kwargs) -> str:
            value = self.get_translation(key=key, language=language, module_name=module_name)
            return value.format(**kwargs)

        return translator_func

    def get_translator_sync(self, language: str, module_name: Optional[str] = None) -> Callable[[str], str]:
        """Return a synchronous translator function for a fixed language.

        Args:
            language: Target language.
            module_name: Optional module context.

        Returns:
            A function `fn(key: str, **kwargs) -> str`.
        """
        def translator_func(key: str, **kwargs) -> str:
            value = self.get_translation(key=key, language=language, module_name=module_name)
            return value.format(**kwargs)
        return translator_func

__init__(bot, global_path, logger)

Initialize the translator and its caches.

Parameters:

Name	Type	Description	Default
bot	ExtendedClient	The extended Discord client.	required
global_path	str	Filesystem path to the global translations directory.	required
logger	Logger	Logger for diagnostics.	required

Source code in utils/Translator.py

def __init__(self, bot: ExtendedClient, global_path: str, logger: logging.Logger):
    """Initialize the translator and its caches.

    Args:
        bot: The extended Discord client.
        global_path: Filesystem path to the global translations directory.
        logger: Logger for diagnostics.
    """
    self.bot = bot
    self.logger = logger
    self.global_path = Path(global_path)
    self.global_translations_cache = {}
    self.module_translation_cache = {}
    self.language_cache = {}

get_language(guild_id) async

Return the configured language for a guild (default 'en').

Normalizes common aliases like "English", "en-US", "pt-br", etc. Results are cached per guild id.

Parameters:

Name	Type	Description	Default
guild_id	str	Discord guild id.	required

Returns:

Type	Description
str	Normalized language code such as "en" or "pt".

Source code in utils/Translator.py

async def get_language(self, guild_id: str) -> str:
    """Return the configured language for a guild (default `'en'`).

    Normalizes common aliases like `"English"`, `"en-US"`, `"pt-br"`, etc.
    Results are cached per guild id.

    Args:
        guild_id: Discord guild id.

    Returns:
        Normalized language code such as `"en"` or `"pt"`.
    """
    guild_id = str(guild_id)
    if guild_id in self.language_cache:
        return self.language_cache[guild_id]

    guild = await self.bot.guild_manager.fetch_or_create(guild_id)
    language = guild.data.get("language", "en")

    if language in ["Inglês", "English", "en", "en-US"]:
        self.language_cache[guild_id] = "en"
        language = "en"
    elif language in ["Português", "pt-br", "Português (Brasileiro)", "pt"]:
        self.language_cache[guild_id] = "pt"
        language = "pt"
    else:
        self.language_cache[guild_id] = language
    return language

get_language_sync(guild_id)

Synchronous variant of get_language using the local cache only.

If the language for guild_id is not cached, returns 'en'.

Parameters:

Name	Type	Description	Default
guild_id	Optional[str]	Guild id (or None).	required

Returns:

Type	Description
str	A normalized language code.

Source code in utils/Translator.py

def get_language_sync(self, guild_id: Optional[str]) -> str:
    """Synchronous variant of `get_language` using the local cache only.

    If the language for `guild_id` is not cached, returns `'en'`.

    Args:
        guild_id: Guild id (or `None`).

    Returns:
        A normalized language code.
    """
    if guild_id and guild_id in self.language_cache:
        guild_id = str(guild_id)
        return self.language_cache[guild_id]
    return "en"

get_translation(key, language, module_name=None)

Return the processed translation for a dot‑separated key.

Parameters:

Name	Type	Description	Default
key	str	Hierarchical key such as "help.commands.ping.description".	required
language	str	Target language code.	required
module_name	Optional[str]	Module name to search in module cache; if omitted, the global cache is used.	None

Returns:

Type	Description
str	The resolved translation string, or key if not found.

Source code in utils/Translator.py

def get_translation(self, key: str, language: str, module_name: Optional[str] = None) -> str:
    """Return the processed translation for a dot‑separated key.

    Args:
        key: Hierarchical key such as `"help.commands.ping.description"`.
        language: Target language code.
        module_name: Module name to search in module cache; if omitted,
            the global cache is used.

    Returns:
        The resolved translation string, or `key` if not found.
    """
    translations = (
        self.module_translation_cache.get(f"{module_name}:{language}", {})
        if module_name
        else self.global_translations_cache.get(language, {})
    )

    keys = key.split(".")
    for k in keys:
        if isinstance(translations, dict):
            translations = translations.get(k)
        else:
            return key  # Retorna a própria chave se não encontrada
    return translations if isinstance(translations, str) else key

get_translator(guild_id, module_name=None) async

Return a callable that resolves translations for a guild asynchronously.

The callable captures the guild's current language and interpolates **kwargs with standard Python str.format.

Parameters:

Name	Type	Description	Default
guild_id	str	Guild id to derive the language from.	required
module_name	Optional[str]	Module name for module‑scoped translations.	None

Returns:

Type	Description
Callable[[str], str]	A function fn(key: str, **kwargs) -> str.

Source code in utils/Translator.py

async def get_translator(self, guild_id: str, module_name: Optional[str] = None) -> Callable[[str], str]:
    """Return a callable that resolves translations for a guild asynchronously.

    The callable captures the guild's current language and interpolates `**kwargs`
    with standard Python `str.format`.

    Args:
        guild_id: Guild id to derive the language from.
        module_name: Module name for module‑scoped translations.

    Returns:
        A function `fn(key: str, **kwargs) -> str`.
    """
    guild_id = str(guild_id)
    language = await self.get_language(guild_id=guild_id)

    def translator_func(key: str, **kwargs) -> str:
        value = self.get_translation(key=key, language=language, module_name=module_name)
        return value.format(**kwargs)

    return translator_func

get_translator_sync(language, module_name=None)

Return a synchronous translator function for a fixed language.

Parameters:

Name	Type	Description	Default
language	str	Target language.	required
module_name	Optional[str]	Optional module context.	None

Returns:

Type	Description
Callable[[str], str]	A function fn(key: str, **kwargs) -> str.

Source code in utils/Translator.py

def get_translator_sync(self, language: str, module_name: Optional[str] = None) -> Callable[[str], str]:
    """Return a synchronous translator function for a fixed language.

    Args:
        language: Target language.
        module_name: Optional module context.

    Returns:
        A function `fn(key: str, **kwargs) -> str`.
    """
    def translator_func(key: str, **kwargs) -> str:
        value = self.get_translation(key=key, language=language, module_name=module_name)
        return value.format(**kwargs)
    return translator_func

refresh_translation_cache() async

Reload global and module translation caches asynchronously.

Scans the global_path for *.json files and each loaded module's translations directory (from its manifest or default "translations").

Source code in utils/Translator.py

async def refresh_translation_cache(self):
    """Reload global and module translation caches asynchronously.

    Scans the `global_path` for `*.json` files and each loaded module's
    translations directory (from its manifest or default `"translations"`).
    """

    available_languages = {file.stem for file in self.global_path.glob("*.json")}
    for language in available_languages:
        path = self.global_path / f"{language}.json"
        if path.exists():
            try:
                async with aiofiles.open(path, "r", encoding="utf-8") as f:
                    content = await f.read()
                    translations = json.loads(content)
                    self.global_translations_cache[language] = self._process_translation(translations)
                    self.logger.info(f"Global translations loaded for language: {language}")
            except json.JSONDecodeError as e:
                self.logger.error(f"Error loading global translations for {language}: {e}")

    # Carregar traduções dos módulos
    for module_name, module in self.bot.modules.items():
        translations_folder = module.data.get("translationsFolder") or "translations"
        translations_path = Path(module.path) / translations_folder
        available_languages = {file.stem for file in translations_path.glob("*.json")}
        for language in available_languages:
            path = translations_path / f"{language}.json"
            if path.exists():
                try:
                    async with aiofiles.open(path, "r", encoding="utf-8") as f:
                        content = await f.read()
                        translations = json.loads(content)
                        processed_translations = self._process_translation(translations, module_name)
                        cache_key = f"{module_name}:{language}"
                        self.module_translation_cache[cache_key] = processed_translations
                        self.logger.info(f"Module translations loaded for {module_name}, language: {language}")
                except json.JSONDecodeError as e:
                    self.logger.error(f"Error loading module translations for {module_name}, language: {language}: {e}")

update_language_cache(guild_id, language)

Update the local language cache for a guild.

Parameters:

Name	Type	Description	Default
guild_id	str	Guild id.	required
language	str	Normalized language code to store.	required

Source code in utils/Translator.py

def update_language_cache(self, guild_id: str, language: str):
    """Update the local language cache for a guild.

    Args:
        guild_id: Guild id.
        language: Normalized language code to store.
    """
    guild_id = str(guild_id)
    self.language_cache[guild_id] = language

utils.Loader

load_commands_from_folder(bot, folder, base_package, logger) async

Dynamically loads commands from a specified folder.

:param bot: The bot instance. :param folder: The Path object for the folder containing command files. :param base_package: The base Python package path to use when loading extensions. :param logger: Logger instance for logging progress and errors.

Source code in utils/Loader.py

async def load_commands_from_folder(bot: commands.Bot, folder: Path, base_package: str, logger: Logger):
    """
    Dynamically loads commands from a specified folder.

    :param bot: The bot instance.
    :param folder: The Path object for the folder containing command files.
    :param base_package: The base Python package path to use when loading extensions.
    :param logger: Logger instance for logging progress and errors.
    """
    if not folder.exists():
        logger.warning(f"Commands folder '{folder}' does not exist.")
        return

    for command_file in folder.glob("*.py"):
        if command_file.stem.startswith("_"):
            # Skip private or special files like __init__.py
            continue

        extension = f"{base_package}.{command_file.stem}"
        try:
            await bot.load_extension(extension)
            logger.info(f"Loaded command: {command_file.stem}")
        except Exception as e:
            logger.error(f"Failed to load command '{command_file.stem}': {e}")

load_translation(module_name, file_name, language, default_language='en')

Carrega o dicionário de traduções baseado no idioma fornecido.

Parameters:

Name	Type	Description	Default
module_name	str	Nome do módulo onde estão as traduções (e.g., "XPSystem").	required
file_name	str	Nome do arquivo de tradução (sem extensão .json).	required
language	str	Idioma preferido da guilda.	required
default_language	str	Idioma padrão caso o preferido não esteja disponível.	'en'

Returns:

Name	Type	Description
Dict	Dict	Dicionário de traduções no idioma solicitado.

Raises:

Type	Description
FileNotFoundError	Se o arquivo de tradução não existir.
ValueError	Se o JSON não puder ser decodificado ou se o idioma não estiver disponível.

Source code in utils/Loader.py

def load_translation(module_name: str, file_name: str, language: str, default_language: str = "en") -> Dict:
    """
    Carrega o dicionário de traduções baseado no idioma fornecido.

    Args:
        module_name (str): Nome do módulo onde estão as traduções (e.g., "XPSystem").
        file_name (str): Nome do arquivo de tradução (sem extensão .json).
        language (str): Idioma preferido da guilda.
        default_language (str): Idioma padrão caso o preferido não esteja disponível.

    Returns:
        Dict: Dicionário de traduções no idioma solicitado.

    Raises:
        FileNotFoundError: Se o arquivo de tradução não existir.
        ValueError: Se o JSON não puder ser decodificado ou se o idioma não estiver disponível.
    """
    logger = logging.getLogger("TranslationLoader")
    translations_path = Path(f"modules/{module_name}/translations/{file_name}.json")

    if not translations_path.exists():
        logger.error(f"Translation file '{translations_path}' not found.")
        raise FileNotFoundError(f"Translation file '{translations_path}' not found.")

    try:
        with open(translations_path, "r", encoding="utf-8") as file:
            translations = json.load(file)
    except json.JSONDecodeError as e:
        logger.error(f"Error decoding JSON file '{translations_path}': {e}")
        raise ValueError(f"Error decoding JSON file '{translations_path}': {e}")

    if language not in translations and default_language not in translations:
        logger.error(f"Language '{language}' and default language '{default_language}' not found in '{translations_path}'.")
        raise ValueError(f"Language '{language}' and default language '{default_language}' not found in '{translations_path}'.")

    logger.debug(f"Successfully loaded translations for language '{language}' from '{translations_path}'.")
    return translations.get(language, translations.get(default_language))

utils.InteractionView

InteractionView

Bases: View, AsyncIOEventEmitter

Interactive Discord UI view that also emits custom events.

This view bridges discord.ui.View with an async event emitter, allowing callers to listen for lifecycle events like "end" (timeout/deletion) while still using standard Discord UI components (buttons/selects).

Features

• Auto timeout and "end" emission with reason "timeout".
• Listens for on_message_delete to end with reason "deleted".
• Component custom_id normalization and namespacing with view_id.
• Ability to clone a view with the same behavior.

Source code in utils/InteractionView.py

class InteractionView(View, AsyncIOEventEmitter):
    """Interactive Discord UI view that also emits custom events.

    This view bridges `discord.ui.View` with an async event emitter, allowing
    callers to listen for lifecycle events like `"end"` (timeout/deletion) while
    still using standard Discord UI components (buttons/selects).

    Features:
      * Auto timeout and `"end"` emission with reason `"timeout"`.
      * Listens for `on_message_delete` to end with reason `"deleted"`.
      * Component `custom_id` normalization and namespacing with `view_id`.
      * Ability to clone a view with the same behavior.
    """
    def __init__(
        self,
        interaction: Interaction,
        channel: TextChannel,
        client: ExtendedClient,
        ephemeral: Optional[bool] = False,
        filter_func: Optional[Callable[[Interaction], bool]] = None,
        timeout: Optional[int] = 60,  # Timeout em segundos
        parent: Optional["InteractionView"] = None,
    ):
        """Create an `InteractionView`.

        Args:
            interaction: The original interaction that spawned this view.
            channel: Text channel for context/logging.
            client: Extended bot client.
            ephemeral: Whether the response should be ephemeral.
            filter_func: Optional predicate to filter accepted interactions.
            timeout: Timeout (seconds) before this view ends automatically.
            parent: Optional parent view if this is a clone.
        """

        View.__init__(self, timeout=timeout)
        AsyncIOEventEmitter.__init__(self)

        self.interaction = interaction
        self.channel = channel
        self.client = client
        self.ephemeral = ephemeral
        self.filter_func = filter_func or (lambda i: True)
        self.parent = parent
        self.msg_id: Optional[str] = str(interaction.message.id) if interaction.message else None
        self.view_id: str = self._generate_random_id()
        self._timeout_task: Optional[asyncio.Task] = None

        self.client.add_listener(self._handle_message_delete, "on_message_delete")

        if self.timeout is not None and self.timeout > 0:
            self.start_timeout()

        self.client.logger.debug(f"InteractionView initialized with view_id: {self.view_id}, message ID: {self.msg_id}, ephemeral: {self.ephemeral}")

    @staticmethod
    def _generate_random_id() -> str:
        """Generate a unique identifier for this view instance.

        Returns:
            A UUID4 string.
        """
        return str(uuid.uuid4())

    async def on_timeout(self):
        """Called by discord.py when the view times out.

        Emits the `"end"` event with reason `"timeout"` and destroys the view.
        """
        self.client.logger.debug(f"View with view_id {self.view_id} has timed out.")
        self.emit("end", "timeout")
        self.destroy("timeout")

    async def _handle_message_delete(self, message: Message):
        """Internal listener to end the view if its message was deleted.

        Args:
            message: The deleted message.
        """
        if message.id == self.msg_id:
            self.client.logger.debug(f"Message with ID {self.msg_id} was deleted, triggering view destruction.")
            self.emit("end", "deleted")
            self.destroy("deleted")

    def start_timeout(self):
        """(Re)start the internal timeout task."""
        if self._timeout_task:
            self._timeout_task.cancel()
        self.client.logger.debug(f"Starting timeout for view with view_id: {self.view_id}")
        self._timeout_task = asyncio.create_task(self._timeout_handler())

    async def _timeout_handler(self):
        """Sleep for `self.timeout` seconds then trigger `on_timeout`."""
        await asyncio.sleep(self.timeout if self.timeout is not None else 0)
        await self.on_timeout()

    async def update(self, **kwargs) -> bool:
        """Update the message associated with this view.

        Keyword Args:
            components: Optional iterable of components (Buttons/Selects) to add.
            Any other arguments accepted by
            `interaction.edit_original_response` or `interaction.response.send_message`.

        Returns:
            True if the update succeeds, False otherwise.
        """
        try:
            components = kwargs.pop("components", [])
            self.client.logger.debug(f"Updating view with components: {components}")
            self.clear_items()
            for component in components:
                component = self._add_custom_id(component)
                self.add_item(component)

            if self.interaction.response.is_done():
                await self.interaction.edit_original_response(view=self, **kwargs)
            else:
                await self.interaction.response.send_message(view=self, **kwargs, ephemeral=self.ephemeral)
            self.client.logger.debug("View updated successfully.")
            return True
        except Exception as e:
            self.client.logger.error(f"Failed to update interaction view: {e}")
            return False

    def _add_custom_id(self, component: Any) -> Any:
        """Ensure the component `custom_id` includes this view's `view_id`.

        This namespaces component identifiers so callbacks can distinguish
        between multiple active views.

        Args:
            component: A Discord UI component instance.

        Returns:
            The same component, potentially with a modified `custom_id`.
        """
        if hasattr(component, "custom_id") and component.custom_id:
            split_id = component.custom_id.split("-")
            if len(split_id) > 1 and "-".join(split_id[1:]) == self.view_id:
                return component 
            component.custom_id = f"{component.custom_id}-{self.view_id}"
            self.client.logger.debug(f"Updated custom_id for component: {component.custom_id} | View Id: {self.view_id}")
        return component

    def normalize_custom_id(self, custom_id: str) -> str:
        """Strip the view_id suffix from a component `custom_id`, if present.

        Args:
            custom_id: The raw custom id from an interaction component.

        Returns:
            The normalized id without the view suffix.
        """
        normalized_id = custom_id.split("-")[0] if "-" in custom_id else custom_id
        self.client.logger.debug(f"Normalized custom_id: {custom_id} -> {normalized_id}")
        return normalized_id

    def clone(self) -> "InteractionView":
        """Create a shallow clone of this view.

        Returns:
            A new `InteractionView` with the same config and `msg_id`.
        """
        self.client.logger.debug(f"Cloning InteractionView with view_id: {self.view_id}")
        cloned_view = InteractionView(
            interaction=self.interaction,
            channel=self.channel,
            client=self.client,
            ephemeral=self.ephemeral,
            filter_func=self.filter_func,
            timeout=int(self.timeout) if self.timeout is not None else None,
            parent=self
        )
        cloned_view.set_msg_id(self.msg_id)
        return cloned_view

    def set_msg_id(self, msg_id: Optional[str]):
        """Attach a message id to this view instance.

        Args:
            msg_id: The Discord message id to associate.
        """
        self.msg_id = msg_id
        self.client.logger.debug(f"Message ID set for view: {msg_id}")

    def destroy(self, reason: Optional[str] = None):
        """Tear down the view and unregister listeners.

        Args:
            reason: Optional reason for diagnostics (e.g., `"timeout"`).
        """
        if self._timeout_task:
            self._timeout_task.cancel()
            self._timeout_task = None

        if self.client.view_registry and self.msg_id in self.client.view_registry:
            del self.client.view_registry[self.msg_id]

        self.emit("end", reason or "destroy")
        self.clear_items() 

        if "on_message_delete" in self._events:
            self.client.remove_listener(self._handle_message_delete, "on_message_delete")

        self.stop()  
        self.client.logger.debug(f"InteractionView with view_id {self.view_id}, {reason}.")

    def set_extra_filter(self, filter_func: Callable[[Interaction], bool]):
        """Set an additional interaction filter predicate.

        Args:
            filter_func: Callable that receives an `Interaction` and returns `True`
                if it should be handled by this view, `False` otherwise.
        """
        self.filter_func = filter_func
        self.client.logger.debug(f"Extra filter function set for InteractionView {self.view_id}")

__init__(interaction, channel, client, ephemeral=False, filter_func=None, timeout=60, parent=None)

Create an InteractionView.

Parameters:

Name	Type	Description	Default
interaction	Interaction	The original interaction that spawned this view.	required
channel	TextChannel	Text channel for context/logging.	required
client	ExtendedClient	Extended bot client.	required
ephemeral	Optional[bool]	Whether the response should be ephemeral.	False
filter_func	Optional[Callable[[Interaction], bool]]	Optional predicate to filter accepted interactions.	None
timeout	Optional[int]	Timeout (seconds) before this view ends automatically.	60
parent	Optional[InteractionView]	Optional parent view if this is a clone.	None

Source code in utils/InteractionView.py

def __init__(
    self,
    interaction: Interaction,
    channel: TextChannel,
    client: ExtendedClient,
    ephemeral: Optional[bool] = False,
    filter_func: Optional[Callable[[Interaction], bool]] = None,
    timeout: Optional[int] = 60,  # Timeout em segundos
    parent: Optional["InteractionView"] = None,
):
    """Create an `InteractionView`.

    Args:
        interaction: The original interaction that spawned this view.
        channel: Text channel for context/logging.
        client: Extended bot client.
        ephemeral: Whether the response should be ephemeral.
        filter_func: Optional predicate to filter accepted interactions.
        timeout: Timeout (seconds) before this view ends automatically.
        parent: Optional parent view if this is a clone.
    """

    View.__init__(self, timeout=timeout)
    AsyncIOEventEmitter.__init__(self)

    self.interaction = interaction
    self.channel = channel
    self.client = client
    self.ephemeral = ephemeral
    self.filter_func = filter_func or (lambda i: True)
    self.parent = parent
    self.msg_id: Optional[str] = str(interaction.message.id) if interaction.message else None
    self.view_id: str = self._generate_random_id()
    self._timeout_task: Optional[asyncio.Task] = None

    self.client.add_listener(self._handle_message_delete, "on_message_delete")

    if self.timeout is not None and self.timeout > 0:
        self.start_timeout()

    self.client.logger.debug(f"InteractionView initialized with view_id: {self.view_id}, message ID: {self.msg_id}, ephemeral: {self.ephemeral}")

clone()

Create a shallow clone of this view.

Returns:

Type	Description
InteractionView	A new InteractionView with the same config and msg_id.

Source code in utils/InteractionView.py

def clone(self) -> "InteractionView":
    """Create a shallow clone of this view.

    Returns:
        A new `InteractionView` with the same config and `msg_id`.
    """
    self.client.logger.debug(f"Cloning InteractionView with view_id: {self.view_id}")
    cloned_view = InteractionView(
        interaction=self.interaction,
        channel=self.channel,
        client=self.client,
        ephemeral=self.ephemeral,
        filter_func=self.filter_func,
        timeout=int(self.timeout) if self.timeout is not None else None,
        parent=self
    )
    cloned_view.set_msg_id(self.msg_id)
    return cloned_view

destroy(reason=None)

Tear down the view and unregister listeners.

Parameters:

Name	Type	Description	Default
reason	Optional[str]	Optional reason for diagnostics (e.g., "timeout").	None

Source code in utils/InteractionView.py

def destroy(self, reason: Optional[str] = None):
    """Tear down the view and unregister listeners.

    Args:
        reason: Optional reason for diagnostics (e.g., `"timeout"`).
    """
    if self._timeout_task:
        self._timeout_task.cancel()
        self._timeout_task = None

    if self.client.view_registry and self.msg_id in self.client.view_registry:
        del self.client.view_registry[self.msg_id]

    self.emit("end", reason or "destroy")
    self.clear_items() 

    if "on_message_delete" in self._events:
        self.client.remove_listener(self._handle_message_delete, "on_message_delete")

    self.stop()  
    self.client.logger.debug(f"InteractionView with view_id {self.view_id}, {reason}.")

normalize_custom_id(custom_id)

Strip the view_id suffix from a component custom_id, if present.

Parameters:

Name	Type	Description	Default
custom_id	str	The raw custom id from an interaction component.	required

Returns:

Type	Description
str	The normalized id without the view suffix.

Source code in utils/InteractionView.py

def normalize_custom_id(self, custom_id: str) -> str:
    """Strip the view_id suffix from a component `custom_id`, if present.

    Args:
        custom_id: The raw custom id from an interaction component.

    Returns:
        The normalized id without the view suffix.
    """
    normalized_id = custom_id.split("-")[0] if "-" in custom_id else custom_id
    self.client.logger.debug(f"Normalized custom_id: {custom_id} -> {normalized_id}")
    return normalized_id

on_timeout() async

Called by discord.py when the view times out.

Emits the "end" event with reason "timeout" and destroys the view.

Source code in utils/InteractionView.py

async def on_timeout(self):
    """Called by discord.py when the view times out.

    Emits the `"end"` event with reason `"timeout"` and destroys the view.
    """
    self.client.logger.debug(f"View with view_id {self.view_id} has timed out.")
    self.emit("end", "timeout")
    self.destroy("timeout")

set_extra_filter(filter_func)

Set an additional interaction filter predicate.

Parameters:

Name	Type	Description	Default
filter_func	Callable[[Interaction], bool]	Callable that receives an Interaction and returns True if it should be handled by this view, False otherwise.	required

Source code in utils/InteractionView.py

def set_extra_filter(self, filter_func: Callable[[Interaction], bool]):
    """Set an additional interaction filter predicate.

    Args:
        filter_func: Callable that receives an `Interaction` and returns `True`
            if it should be handled by this view, `False` otherwise.
    """
    self.filter_func = filter_func
    self.client.logger.debug(f"Extra filter function set for InteractionView {self.view_id}")

set_msg_id(msg_id)

Attach a message id to this view instance.

Parameters:

Name	Type	Description	Default
msg_id	Optional[str]	The Discord message id to associate.	required

Source code in utils/InteractionView.py

def set_msg_id(self, msg_id: Optional[str]):
    """Attach a message id to this view instance.

    Args:
        msg_id: The Discord message id to associate.
    """
    self.msg_id = msg_id
    self.client.logger.debug(f"Message ID set for view: {msg_id}")

start_timeout()

(Re)start the internal timeout task.

Source code in utils/InteractionView.py

def start_timeout(self):
    """(Re)start the internal timeout task."""
    if self._timeout_task:
        self._timeout_task.cancel()
    self.client.logger.debug(f"Starting timeout for view with view_id: {self.view_id}")
    self._timeout_task = asyncio.create_task(self._timeout_handler())

update(**kwargs) async

Update the message associated with this view.

Other Parameters:

Name	Type	Description
components		Optional iterable of components (Buttons/Selects) to add.

Returns:

Type	Description
bool	True if the update succeeds, False otherwise.

Source code in utils/InteractionView.py

async def update(self, **kwargs) -> bool:
    """Update the message associated with this view.

    Keyword Args:
        components: Optional iterable of components (Buttons/Selects) to add.
        Any other arguments accepted by
        `interaction.edit_original_response` or `interaction.response.send_message`.

    Returns:
        True if the update succeeds, False otherwise.
    """
    try:
        components = kwargs.pop("components", [])
        self.client.logger.debug(f"Updating view with components: {components}")
        self.clear_items()
        for component in components:
            component = self._add_custom_id(component)
            self.add_item(component)

        if self.interaction.response.is_done():
            await self.interaction.edit_original_response(view=self, **kwargs)
        else:
            await self.interaction.response.send_message(view=self, **kwargs, ephemeral=self.ephemeral)
        self.client.logger.debug("View updated successfully.")
        return True
    except Exception as e:
        self.client.logger.error(f"Failed to update interaction view: {e}")
        return False

utils.MessageView

MessageView

Bases: AsyncIOEventEmitter

Source code in utils/MessageView.py

class MessageView(AsyncIOEventEmitter):
    def __init__(
        self,
        message: Message,
        channel: TextChannel,
        client: ExtendedClient,
        filter_func: Optional[Callable[[Interaction], bool]] = None,
        timeout: Optional[int] = 60000,
    ):
        """
        MessageView is a utility for managing interactive views tied to messages in a Discord bot.
        """
        super().__init__()
        self.message = message
        self.channel = channel
        self.client = client
        self.msg_id = message.id
        self.filter_func = filter_func or (lambda _: True)
        self.timeout = timeout
        self.view_id = str(uuid.uuid4())
        self._timeout_task: Optional[asyncio.Task] = None

        # Register event listeners
        self.client.on("interaction_create", self._handle_interaction)
        self.client.on("message_delete", self._handle_message_delete)

        # Start timeout task if needed
        if self.timeout > 0:
            self._start_timeout()

    def _start_timeout(self):
        if self._timeout_task:
            self._timeout_task.cancel()
        self._timeout_task = asyncio.create_task(self._timeout_handler())

    async def _timeout_handler(self):
        await asyncio.sleep(self.timeout / 1000)
        self.destroy("timeout")

    async def _handle_interaction(self, interaction: Interaction):
        if interaction.message and interaction.message.id == self.msg_id:
            split_id = interaction.custom_id.split("-")
            event_id = split_id.pop(0)
            view_id = split_id.pop(-1)

            if view_id == self.view_id and self.filter_func(interaction):
                self.emit(event_id, interaction)
                self.emit("any", interaction)

    async def _handle_message_delete(self, message: Message):
        if message.id == self.msg_id:
            self.destroy("deleted")

    async def update(self, view: MessageViewUpdate) -> bool:
        """
        Update the view with new content, embeds, or components.
        """
        try:
            if "components" in view:
                view["components"] = self._add_random_id_to_buttons(view["components"])
            await self.message.edit(**view)
            return True
        except Exception as e:
            self.client.logger.error(f"Failed to update message view: {e}")
            return False

    def clone(self) -> "MessageView":
        """
        Create a cloned instance of this MessageView.
        """
        return MessageView(
            message=self.message,
            channel=self.channel,
            client=self.client,
            filter_func=self.filter_func,
            timeout=self.timeout,
        )

    def destroy(self, reason: Optional[str] = None):
        """
        Destroy this view and clean up listeners.
        """
        if self._timeout_task:
            self._timeout_task.cancel()
            self._timeout_task = None

        self.emit("end", reason or "destroy")
        self.remove_all_listeners()

        self.client.off("interaction_create", self._handle_interaction)
        self.client.off("message_delete", self._handle_message_delete)

    def _add_random_id_to_buttons(self, rows: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """
        Adds a random ID to buttons for unique identification.
        """
        for row in rows:
            for component in row.get("components", []):
                custom_id = component.get("custom_id", "")
                split_id = custom_id.split("-")
                if split_id[-1] != self.view_id:
                    component["custom_id"] = f"{custom_id}-{self.view_id}"
        return rows

__init__(message, channel, client, filter_func=None, timeout=60000)

MessageView is a utility for managing interactive views tied to messages in a Discord bot.

Source code in utils/MessageView.py

def __init__(
    self,
    message: Message,
    channel: TextChannel,
    client: ExtendedClient,
    filter_func: Optional[Callable[[Interaction], bool]] = None,
    timeout: Optional[int] = 60000,
):
    """
    MessageView is a utility for managing interactive views tied to messages in a Discord bot.
    """
    super().__init__()
    self.message = message
    self.channel = channel
    self.client = client
    self.msg_id = message.id
    self.filter_func = filter_func or (lambda _: True)
    self.timeout = timeout
    self.view_id = str(uuid.uuid4())
    self._timeout_task: Optional[asyncio.Task] = None

    # Register event listeners
    self.client.on("interaction_create", self._handle_interaction)
    self.client.on("message_delete", self._handle_message_delete)

    # Start timeout task if needed
    if self.timeout > 0:
        self._start_timeout()

clone()

Create a cloned instance of this MessageView.

Source code in utils/MessageView.py

def clone(self) -> "MessageView":
    """
    Create a cloned instance of this MessageView.
    """
    return MessageView(
        message=self.message,
        channel=self.channel,
        client=self.client,
        filter_func=self.filter_func,
        timeout=self.timeout,
    )

destroy(reason=None)

Destroy this view and clean up listeners.

Source code in utils/MessageView.py

def destroy(self, reason: Optional[str] = None):
    """
    Destroy this view and clean up listeners.
    """
    if self._timeout_task:
        self._timeout_task.cancel()
        self._timeout_task = None

    self.emit("end", reason or "destroy")
    self.remove_all_listeners()

    self.client.off("interaction_create", self._handle_interaction)
    self.client.off("message_delete", self._handle_message_delete)

update(view) async

Update the view with new content, embeds, or components.

Source code in utils/MessageView.py

async def update(self, view: MessageViewUpdate) -> bool:
    """
    Update the view with new content, embeds, or components.
    """
    try:
        if "components" in view:
            view["components"] = self._add_random_id_to_buttons(view["components"])
        await self.message.edit(**view)
        return True
    except Exception as e:
        self.client.logger.error(f"Failed to update message view: {e}")
        return False

create_view(channel, client, view, filter_func=None) async

Create a MessageView tied to a newly sent message.

Source code in utils/MessageView.py

async def create_view(
    channel: TextChannel, client: ExtendedClient, view: MessageViewUpdate, filter_func: Optional[Callable[[Interaction], bool]] = None
) -> MessageView:
    """
    Create a MessageView tied to a newly sent message.
    """
    message = await channel.send(**view)
    return MessageView(message, channel, client, filter_func)

create_view_from_interaction(interaction, client, view, filter_func=None) async

Create a MessageView tied to an interaction's response.

Source code in utils/MessageView.py

async def create_view_from_interaction(
    interaction: Interaction, client: ExtendedClient, view: MessageViewUpdate, filter_func: Optional[Callable[[Interaction], bool]] = None
) -> MessageView:
    """
    Create a MessageView tied to an interaction's response.
    """
    if not interaction.channel:
        raise ValueError("Interaction channel not found")

    response_message = await interaction.response.send_message(**view)
    return MessageView(response_message, interaction.channel, client, filter_func)

create_view_from_message(message, client, filter_func=None) async

Create a MessageView tied to an existing message.

Source code in utils/MessageView.py

async def create_view_from_message(
    message: Message, client: ExtendedClient, filter_func: Optional[Callable[[Interaction], bool]] = None
) -> MessageView:
    """
    Create a MessageView tied to an existing message.
    """
    return MessageView(message, message.channel, client, filter_func)

utils.ViewRouter

ViewRouter

Bases: AsyncIOEventEmitter

A class to manage a navigation stack for interactive views.

Source code in utils/ViewRouter.py

class ViewRouter(AsyncIOEventEmitter):
    """A class to manage a navigation stack for interactive views."""

    def __init__(self, logger, view):
        """
        Initialize the ViewRouter.

        Args:
            logger: Logger instance for logging.
            view: The current active view (e.g., InteractionView or similar).
        """
        super().__init__()
        self.logger = logger.getChild("ViewRouter")
        self.stack: List[Dict[str, Any]] = []
        self.view = view
        self.forced_rows: List[ActionRow] = []

        # Forward events from the managed view to this router
        self._forward_events(view, self)

        # Handle "returnPage" event to pop the view
        self.on("returnPage", self.pop)

    def _forward_events(self, forwarder, forwarded):
        """Forward events from one EventEmitter to another."""
        original_emit = forwarder.emit

        def new_emit(event_name: str, *args):
            forwarded.emit(event_name, *args)
            return original_emit(event_name, *args)

        forwarder.emit = new_emit

    def set_view(self, view):
        """Set the active view managed by the router."""
        self.view = view

    def set_rows(self, rows: List[ActionRow]):
        """Set persistent rows to append to every page."""
        self.forced_rows = rows

    async def push(self, update: Dict[str, Any]) -> str:
        """
        Push a new page onto the navigation stack.

        Args:
            update: A dictionary containing the new view's content and components.

        Returns:
            A unique ID for the pushed page.
        """
        page_id = str(uuid4())
        self.stack.append({"style": update, "id": page_id})

        # Append forced rows to components
        if "components" in update:
            update["components"] += self.forced_rows
        else:
            update["components"] = self.forced_rows

        await self.view.update(update)
        return page_id

    async def pop(self):
        """
        Pop the current page from the navigation stack.

        Returns:
            The unique ID of the popped page, or None if the stack is empty.
        """
        if not self.stack:
            return None

        page = self.stack.pop()
        await self.view.update(page["style"])
        return page["id"]

    def clear_stack(self):
        """Clear the navigation stack."""
        self.stack = []

    async def update(self, update: Dict[str, Any]) -> bool:
        """
        Update the current view without modifying the stack.

        Args:
            update: A dictionary containing the updated view's content and components.

        Returns:
            True if the update was successful, False otherwise.
        """
        return await self.view.update(update)

    async def destroy(self):
        """Destroy the view and clean up resources."""
        await self.view.destroy()

__init__(logger, view)

Initialize the ViewRouter.

Parameters:

Name	Type	Description	Default
logger		Logger instance for logging.	required
view		The current active view (e.g., InteractionView or similar).	required

Source code in utils/ViewRouter.py

def __init__(self, logger, view):
    """
    Initialize the ViewRouter.

    Args:
        logger: Logger instance for logging.
        view: The current active view (e.g., InteractionView or similar).
    """
    super().__init__()
    self.logger = logger.getChild("ViewRouter")
    self.stack: List[Dict[str, Any]] = []
    self.view = view
    self.forced_rows: List[ActionRow] = []

    # Forward events from the managed view to this router
    self._forward_events(view, self)

    # Handle "returnPage" event to pop the view
    self.on("returnPage", self.pop)

clear_stack()

Clear the navigation stack.

Source code in utils/ViewRouter.py

def clear_stack(self):
    """Clear the navigation stack."""
    self.stack = []

destroy() async

Destroy the view and clean up resources.

Source code in utils/ViewRouter.py

async def destroy(self):
    """Destroy the view and clean up resources."""
    await self.view.destroy()

pop() async

Pop the current page from the navigation stack.

Returns:

Type	Description
	The unique ID of the popped page, or None if the stack is empty.

Source code in utils/ViewRouter.py

async def pop(self):
    """
    Pop the current page from the navigation stack.

    Returns:
        The unique ID of the popped page, or None if the stack is empty.
    """
    if not self.stack:
        return None

    page = self.stack.pop()
    await self.view.update(page["style"])
    return page["id"]

push(update) async

Push a new page onto the navigation stack.

Parameters:

Name	Type	Description	Default
update	Dict[str, Any]	A dictionary containing the new view's content and components.	required

Returns:

Type	Description
str	A unique ID for the pushed page.

Source code in utils/ViewRouter.py

async def push(self, update: Dict[str, Any]) -> str:
    """
    Push a new page onto the navigation stack.

    Args:
        update: A dictionary containing the new view's content and components.

    Returns:
        A unique ID for the pushed page.
    """
    page_id = str(uuid4())
    self.stack.append({"style": update, "id": page_id})

    # Append forced rows to components
    if "components" in update:
        update["components"] += self.forced_rows
    else:
        update["components"] = self.forced_rows

    await self.view.update(update)
    return page_id

set_rows(rows)

Set persistent rows to append to every page.

Source code in utils/ViewRouter.py

def set_rows(self, rows: List[ActionRow]):
    """Set persistent rows to append to every page."""
    self.forced_rows = rows

set_view(view)

Set the active view managed by the router.

Source code in utils/ViewRouter.py

def set_view(self, view):
    """Set the active view managed by the router."""
    self.view = view

update(update) async

Update the current view without modifying the stack.

Parameters:

Name	Type	Description	Default
update	Dict[str, Any]	A dictionary containing the updated view's content and components.	required

Returns:

Type	Description
bool	True if the update was successful, False otherwise.

Source code in utils/ViewRouter.py

async def update(self, update: Dict[str, Any]) -> bool:
    """
    Update the current view without modifying the stack.

    Args:
        update: A dictionary containing the updated view's content and components.

    Returns:
        True if the update was successful, False otherwise.
    """
    return await self.view.update(update)

utils.components.PaginatorComponent

PaginatorComponent

Bases: AsyncIOEventEmitter

Source code in utils/components/PaginatorComponent.py

class PaginatorComponent(AsyncIOEventEmitter):
    def __init__(
        self,
        view: AnyView,
        pages: List[Page],
        flags: Optional[List[int]] = None,
        page_update: Optional[PageUpdateFn] = None,
        control_style: Optional[ControlStyle] = None,
    ):
        super().__init__()
        self.view = view
        self.pages = pages
        self.flags = flags or [PaginatorFlags.WRAP]
        self.page_update = page_update
        self.current_page = 0
        self.total_pages = len(pages)

        # Configurar botões traduzidos
        self.buttons = self._setup_translated_buttons(control_style)

        # Configurar eventos de interação
        self.view.on("nextPage", self.next_page_interaction)
        self.view.on("previousPage", self.previous_page_interaction)

    def _setup_translated_buttons(self, control_style: Optional[ControlStyle]) -> Dict[str, Button]:
        """
        Configura os botões com traduções globais.
        """
        # Obter o idioma e as traduções
        client = self.view.client
        guild_id = self.view.interaction.guild.id if hasattr(self.view.interaction, "guild") else None
        language = client.translator.get_language_sync(guild_id) if guild_id else "en"
        translate = client.translator.get_global(language)

        # Configurar botões com rótulos traduzidos
        buttons = {
            "previous": Button(
                label=translate("paginator.previous"),
                style=ButtonStyle.primary,
                custom_id="previousPage",
            ),
            "select": Button(
                label=translate("paginator.select"),
                style=ButtonStyle.primary,
                custom_id="selectPage",
            ),
            "next": Button(
                label=translate("paginator.next"),
                style=ButtonStyle.primary,
                custom_id="nextPage",
            ),
        }

        # Substituir botões padrão por estilos personalizados, se fornecidos
        if control_style:
            buttons.update(control_style)

        return buttons



    def add_pagination_controls(self, page_data: Dict[str, Any]) -> Dict[str, Any]:
        """
        Adiciona controles de paginação ao layout da página.
        """
        if page_data.get("hasControls", False):
            return page_data

        # Adicionar botões diretamente como objetos Button
        row = View()
        row.add_item(self.buttons["previous"])
        row.add_item(self.buttons["select"])
        row.add_item(self.buttons["next"])

        # Adicionar o layout ao campo 'components'
        components = page_data.get("components", [])
        components.append(row)
        page_data["components"] = components
        page_data["hasControls"] = True

        return page_data


    async def update_view(self, page: int):
        """
        Atualiza a view com a página especificada.
        """
        page_data = self.page_update(self.pages[page]) if self.page_update else self.pages[page]
        page_data = self.add_pagination_controls(page_data)
        await self.view.update(page_data)

    async def next_page_interaction(self, interaction: Interaction):
        """
        Lida com a interação do botão "Próximo".
        """
        client = self.view.client
        guild_id = self.view.interaction.guild.id if hasattr(self.view.interaction, "guild") else None
        language = client.translator.get_language_sync(guild_id) if guild_id else "en"
        translate = client.translator.get_global(language)

        if self.current_page + 1 < self.total_pages:
            self.current_page += 1
            await interaction.response.defer_update()
            await self.update_view(self.current_page)
        elif PaginatorFlags.WRAP in self.flags:
            self.current_page = 0
            await interaction.response.defer_update()
            await self.update_view(self.current_page)
        else:
            await interaction.response.send_message(translate("paginator.last_page"), ephemeral=True)

    async def previous_page_interaction(self, interaction: Interaction):
        """
        Lida com a interação do botão "Anterior".
        """
        client = self.view.client
        guild_id = self.view.interaction.guild.id if hasattr(self.view.interaction, "guild") else None
        language = client.translator.get_language_sync(guild_id) if guild_id else "en"
        translate = client.translator.get_global(language)

        if self.current_page > 0:
            self.current_page -= 1
            await interaction.response.defer_update()
            await self.update_view(self.current_page)
        elif PaginatorFlags.WRAP in self.flags:
            self.current_page = self.total_pages - 1
            await interaction.response.defer_update()
            await self.update_view(self.current_page)
        else:
            await interaction.response.send_message(translate("paginator.first_page"), ephemeral=True)

    def set_update_function(self, fn: PageUpdateFn):
        """
        Define a função para atualizar as páginas.
        """
        self.page_update = fn

    async def next_page(self):
        """
        Avança para a próxima página programaticamente.
        """
        if self.current_page + 1 < self.total_pages:
            self.current_page += 1
            await self.update_view(self.current_page)

    async def previous_page(self):
        """
        Retorna à página anterior programaticamente.
        """
        if self.current_page > 0:
            self.current_page -= 1
            await self.update_view(self.current_page)

    async def set_page(self, page: int):
        """
        Define uma página específica como a atual.
        """
        if 0 <= page < self.total_pages:
            self.current_page = page
            await self.update_view(self.current_page)

    async def init(self):
        """
        Inicializa o paginator na primeira página.
        """
        await self.update_view(self.current_page)
        self.view.refresh_timeout()

add_pagination_controls(page_data)

Adiciona controles de paginação ao layout da página.

Source code in utils/components/PaginatorComponent.py

def add_pagination_controls(self, page_data: Dict[str, Any]) -> Dict[str, Any]:
    """
    Adiciona controles de paginação ao layout da página.
    """
    if page_data.get("hasControls", False):
        return page_data

    # Adicionar botões diretamente como objetos Button
    row = View()
    row.add_item(self.buttons["previous"])
    row.add_item(self.buttons["select"])
    row.add_item(self.buttons["next"])

    # Adicionar o layout ao campo 'components'
    components = page_data.get("components", [])
    components.append(row)
    page_data["components"] = components
    page_data["hasControls"] = True

    return page_data

init() async

Inicializa o paginator na primeira página.

Source code in utils/components/PaginatorComponent.py

async def init(self):
    """
    Inicializa o paginator na primeira página.
    """
    await self.update_view(self.current_page)
    self.view.refresh_timeout()

next_page() async

Avança para a próxima página programaticamente.

Source code in utils/components/PaginatorComponent.py

async def next_page(self):
    """
    Avança para a próxima página programaticamente.
    """
    if self.current_page + 1 < self.total_pages:
        self.current_page += 1
        await self.update_view(self.current_page)

next_page_interaction(interaction) async

Lida com a interação do botão "Próximo".

Source code in utils/components/PaginatorComponent.py

async def next_page_interaction(self, interaction: Interaction):
    """
    Lida com a interação do botão "Próximo".
    """
    client = self.view.client
    guild_id = self.view.interaction.guild.id if hasattr(self.view.interaction, "guild") else None
    language = client.translator.get_language_sync(guild_id) if guild_id else "en"
    translate = client.translator.get_global(language)

    if self.current_page + 1 < self.total_pages:
        self.current_page += 1
        await interaction.response.defer_update()
        await self.update_view(self.current_page)
    elif PaginatorFlags.WRAP in self.flags:
        self.current_page = 0
        await interaction.response.defer_update()
        await self.update_view(self.current_page)
    else:
        await interaction.response.send_message(translate("paginator.last_page"), ephemeral=True)

previous_page() async

Retorna à página anterior programaticamente.

Source code in utils/components/PaginatorComponent.py

async def previous_page(self):
    """
    Retorna à página anterior programaticamente.
    """
    if self.current_page > 0:
        self.current_page -= 1
        await self.update_view(self.current_page)

previous_page_interaction(interaction) async

Lida com a interação do botão "Anterior".

Source code in utils/components/PaginatorComponent.py

async def previous_page_interaction(self, interaction: Interaction):
    """
    Lida com a interação do botão "Anterior".
    """
    client = self.view.client
    guild_id = self.view.interaction.guild.id if hasattr(self.view.interaction, "guild") else None
    language = client.translator.get_language_sync(guild_id) if guild_id else "en"
    translate = client.translator.get_global(language)

    if self.current_page > 0:
        self.current_page -= 1
        await interaction.response.defer_update()
        await self.update_view(self.current_page)
    elif PaginatorFlags.WRAP in self.flags:
        self.current_page = self.total_pages - 1
        await interaction.response.defer_update()
        await self.update_view(self.current_page)
    else:
        await interaction.response.send_message(translate("paginator.first_page"), ephemeral=True)

set_page(page) async

Define uma página específica como a atual.

Source code in utils/components/PaginatorComponent.py

async def set_page(self, page: int):
    """
    Define uma página específica como a atual.
    """
    if 0 <= page < self.total_pages:
        self.current_page = page
        await self.update_view(self.current_page)

set_update_function(fn)

Define a função para atualizar as páginas.

Source code in utils/components/PaginatorComponent.py

def set_update_function(self, fn: PageUpdateFn):
    """
    Define a função para atualizar as páginas.
    """
    self.page_update = fn

update_view(page) async

Atualiza a view com a página especificada.

Source code in utils/components/PaginatorComponent.py

async def update_view(self, page: int):
    """
    Atualiza a view com a página especificada.
    """
    page_data = self.page_update(self.pages[page]) if self.page_update else self.pages[page]
    page_data = self.add_pagination_controls(page_data)
    await self.view.update(page_data)

create_paginator(view, pages, flags=None, control_style=None) async

Cria e retorna um componente de paginação.

Source code in utils/components/PaginatorComponent.py

async def create_paginator(
    view: AnyView,
    pages: List[Page],
    flags: Optional[List[int]] = None,
    control_style: Optional[ControlStyle] = None,
) -> PaginatorComponent:
    """
    Cria e retorna um componente de paginação.
    """
    paginator = PaginatorComponent(view, pages, flags, None, control_style)
    if flags and PaginatorFlags.AUTO_INIT in flags:
        await paginator.init()
    return paginator

utils.components.EmbedCreatorComponent

embed_creator(view, filter_func, should_complete=True, data=None) async

Cria um embed interativo para ser configurado pelo usuário, com suporte a traduções globais.

Source code in utils/components/EmbedCreatorComponent.py

async def embed_creator(
    view: AnyView,
    filter_func: Callable[[Message], bool],
    should_complete: bool = True,
    data: Optional[Embed] = None,
) -> Embed:
    """
    Cria um embed interativo para ser configurado pelo usuário, com suporte a traduções globais.
    """
    client = view.client

    # Determinar o idioma
    guild_id = (
        view.interaction.guild.id if isinstance(view, InteractionView) and view.interaction.guild else
        view.message.guild.id if isinstance(view, MessageView) and view.message.guild else
        None
    )
    language = await client.translator.get_language(guild_id) if guild_id else "en"
    translate = client.translator.get_global(language)

    # Criar o embed padrão
    default_embed = Embed(
        title=translate("embed_creator.default_title"),
        description=translate("embed_creator.default_description"),
        color=0xFFFFFF,
    )
    embed = data or default_embed

    # Configurar os botões
    rows = [
        {"custom_id": "title", "label": translate("embed_creator.buttons.title")},
        {"custom_id": "description", "label": translate("embed_creator.buttons.description")},
        {"custom_id": "color", "label": translate("embed_creator.buttons.color")},
        {"custom_id": "image", "label": translate("embed_creator.buttons.image")},
        {"custom_id": "thumbnail", "label": translate("embed_creator.buttons.thumbnail")},
    ]

    await view.update(
        {
            "embeds": [embed.to_dict()],
            "components": [{"type": 1, "components": rows}],
        }
    )

    # Função para gerenciar eventos
    async def handle_event(event_name: str, prompt_key: str, update_func: Callable[[str], None]):
        question_embed = Embed(
            title=translate("embed_creator.editing"),
            description=translate(prompt_key),
            color=0xFFFFFF,
        )

        await view.update({"embeds": [question_embed.to_dict()], "components": []})
        user_input = await wait_for_user_input(view, {}, 30 * SECOND, filter_func)

        if user_input:
            update_func(user_input)
            await view.update(
                {"embeds": [embed.to_dict()], "components": [{"type": 1, "components": rows}]}
            )
        else:
            error_embed = Embed(
                title=translate("embed_creator.error"),
                description=translate("embed_creator.error_timeout"),
                color=0xFF0000,
            )
            await view.update({"embeds": [error_embed.to_dict()]})

    # Eventos
    async def on_title(_interaction):
        await handle_event(
            "title",
            "embed_creator.prompts.title",
            lambda input: embed.update(title=input),
        )

    async def on_description(_interaction):
        await handle_event(
            "description",
            "embed_creator.prompts.description",
            lambda input: embed.update(description=input),
        )

    # Registrar eventos no view
    view.on("title", on_title)
    view.on("description", on_description)

    if should_complete:
        view.on(
            "finish",
            lambda _interaction: asyncio.create_task(view.destroy("Finalizado")),
        )

    return embed

wait_for_user_input(view, new_view, expiry, filter_func, options=None) async

Aguarda a entrada do usuário em uma mensagem ou interação.

Source code in utils/components/EmbedCreatorComponent.py

async def wait_for_user_input(
    view: AnyView,
    new_view: MessageViewUpdate,
    expiry: int,
    filter_func: Callable[[Message], bool],
    options: Optional[dict] = None,
) -> Union[str, None]:
    """
    Aguarda a entrada do usuário em uma mensagem ou interação.
    """
    options = options or {"deleteCollected": True}
    await view.update(new_view)
    channel: TextChannel = view.channel

    def check(message: Message):
        return filter_func(message)

    try:
        message = await view.client.wait_for(
            "message",
            check=check,
            timeout=expiry / SECOND,
        )

        if options.get("deleteCollected"):
            await message.delete()

        if message.content:
            return message.content

        if message.attachments:
            return message.attachments[0].url

    except asyncio.TimeoutError:
        return None