tux.utils.emoji ¶

async def init(self) -> bool:
    """Initializes the emoji cache by fetching application emojis.

    Ensures the cache reflects the current state of application emojis on Discord.
    This method is locked to prevent concurrent initialization attempts.

    Returns
    -------
    bool
        True if initialization was successful or already done, False otherwise.
    """

    async with self._init_lock:
        if self._initialized:
            logger.debug("Emoji cache already initialized.")
            return True

        logger.info("Initializing emoji manager and cache...")

        try:
            app_emojis = await self.bot.fetch_application_emojis()
            self.cache = {emoji.name: emoji for emoji in app_emojis if _is_valid_emoji_name(emoji.name)}

            logger.info(f"Initialized emoji cache with {len(self.cache)} emojis.")
            self._initialized = True

        except discord.HTTPException as e:
            logger.error(f"Failed to fetch application emojis during init: {e}")
            self._initialized = False
            return False
        except Exception:
            logger.exception("Unexpected error during emoji cache initialization.")
            self._initialized = False
            return False

        else:
            return True

def get(self, name: str) -> discord.Emoji | None:
    """Retrieves an emoji from the cache.

    Ensures initialization before attempting retrieval.

    Parameters
    ----------
    name : str
        The name of the emoji to retrieve.

    Returns
    -------
    discord.Emoji | None
        The discord.Emoji object if found, None otherwise.
    """

    if not self._initialized:
        logger.warning("Attempted to get emoji before cache initialization. Call await manager.init() first.")

        # Avoid deadlocks: Do not call init() here directly.
        # Rely on the initial setup_hook call.
        return None

    return self.cache.get(name)

async def _create_discord_emoji(self, name: str, image_bytes: bytes) -> discord.Emoji | None:
    """Internal helper to create a Discord emoji with error handling and delay.

    Parameters
    ----------
    name : str
        The name of the emoji to create.
    image_bytes : bytes
        The image bytes of the emoji to create.

    Returns
    -------
    discord.Emoji | None
        The newly created emoji if successful, otherwise None.
    """

    if not _is_valid_emoji_name(name):
        logger.error(f"Attempted to create emoji with invalid name: '{name}'")
        return None

    try:
        await asyncio.sleep(self.create_delay)
        emoji = await self.bot.create_application_emoji(name=name, image=image_bytes)
        self.cache[name] = emoji  # Update cache immediately
        logger.info(f"Successfully created emoji '{name}'. ID: {emoji.id}")
        return emoji  # noqa: TRY300

    except discord.HTTPException as e:
        logger.error(f"Failed to create emoji '{name}': {e}")
    except ValueError as e:
        logger.error(f"Invalid value for creating emoji '{name}': {e}")
    except Exception as e:
        logger.exception(f"An unexpected error occurred creating emoji '{name}': {e}")

    return None

async def _process_emoji_file(self, file_path: Path) -> tuple[discord.Emoji | None, Path | None]:
    """Attempts to process a single emoji file.

    Parameters
    ----------
    file_path : Path
        The path to the emoji file to process

    Returns
    -------
    tuple[discord.Emoji | None, Path | None]
        A tuple where the first element is the newly created emoji (if created)
        and the second element is the file_path if processing failed or was skipped.
    """
    if not file_path.is_file():
        logger.trace(f"Skipping non-file item: {file_path.name}")
        return None, file_path

    emoji_name = file_path.stem

    if not _is_valid_emoji_name(emoji_name):
        logger.warning(f"Skipping file with invalid potential emoji name: {file_path.name}")
        return None, file_path

    if self.get(emoji_name):
        logger.trace(f"Emoji '{emoji_name}' already exists, skipping.")
        return None, file_path

    logger.debug(f"Emoji '{emoji_name}' not found in cache, attempting to create from {file_path.name}.")

    if img_bytes := _read_emoji_file(file_path):
        new_emoji = await self._create_discord_emoji(emoji_name, img_bytes)
        if new_emoji:
            return new_emoji, None

    return None, file_path  # Failed creation or read

async def sync_emojis(self) -> tuple[list[discord.Emoji], list[Path]]:
    """Synchronizes emojis from the local assets directory to the application.

    Ensures the cache is initialized, then iterates through local emoji files.
    If an emoji with the same name doesn't exist in the cache, it attempts to create it.

    Returns
    -------
    tuple[list[discord.Emoji], list[Path]]
        A tuple containing:
        - A list of successfully created discord.Emoji objects.
        - A list of file paths for emojis that already existed or failed.
    """

    if not await self._ensure_initialized():
        logger.error("Cannot sync emojis: Cache initialization failed.")
        # Attempt to list files anyway for the return value

        with contextlib.suppress(Exception):
            return [], list(self.emojis_path.iterdir())
        return [], []

    logger.info(f"Starting emoji synchronization from {self.emojis_path}...")

    duplicates_or_failed: list[Path] = []
    created_emojis: list[discord.Emoji] = []

    try:
        files_to_process = list(self.emojis_path.iterdir())
    except OSError as e:
        logger.error(f"Failed to list files in emoji directory {self.emojis_path}: {e}")
        return [], []

    if not files_to_process:
        logger.warning(f"No files found in emoji directory: {self.emojis_path}")
        return [], []

    for file_path in files_to_process:
        emoji, failed_file = await self._process_emoji_file(file_path)
        if emoji:
            created_emojis.append(emoji)
        elif failed_file:
            duplicates_or_failed.append(failed_file)

    logger.info(
        f"Emoji synchronization finished. "
        f"Created: {len(created_emojis)}, Duplicates/Skipped/Failed: {len(duplicates_or_failed)}.",
    )

    return created_emojis, duplicates_or_failed

async def _ensure_initialized(self) -> bool:
    """Internal helper: Checks if cache is initialized, logs warning if not."""
    if self._initialized:
        return True
    logger.warning("Operation called before cache was initialized. Call await manager.init() first.")
    # Attempting init() again might lead to issues/deadlocks depending on context.
    # Force initialization in setup_hook.
    return False

async def _delete_discord_emoji(self, name: str) -> bool:
    """Internal helper: Deletes an existing Discord emoji by name and updates cache.

    Parameters
    ----------
    name : str
        The name of the emoji to delete.

    Returns
    -------
    bool
        True if the emoji was deleted, False otherwise.
    """

    existing_emoji = self.get(name)
    if not existing_emoji:
        logger.info(f"No existing emoji '{name}' found in cache. Skipping deletion.")
        return False  # Indicate no deletion occurred

    logger.debug(f"Attempting deletion of application emoji '{name}'...")
    deleted_on_discord = False

    try:
        await existing_emoji.delete()
        logger.info(f"Successfully deleted existing application emoji '{name}'.")
        deleted_on_discord = True

    except discord.NotFound:
        logger.warning(f"Emoji '{name}' was in cache but not found on Discord for deletion.")
    except discord.Forbidden:
        logger.error(f"Missing permissions to delete application emoji '{name}'.")
    except discord.HTTPException as e:
        logger.error(f"Failed to delete application emoji '{name}': {e}")
    except Exception as e:
        logger.exception(f"An unexpected error occurred deleting emoji '{name}': {e}")

    finally:
        # Always remove from cache if it was found initially
        if self.cache.pop(name, None):
            logger.debug(f"Removed '{name}' from cache.")

    return deleted_on_discord

async def resync_emoji(self, name: str) -> discord.Emoji | None:
    """Resyncs a specific emoji: Deletes existing, finds local file, creates new.

    Parameters
    ----------
    name : str
        The name of the emoji to resync.

    Returns
    -------
    Optional[discord.Emoji]
        The newly created emoji if successful, otherwise None.
    """

    logger.info(f"Starting resync process for emoji: '{name}'...")

    if not await self._ensure_initialized():
        return None  # Stop if initialization failed

    # Step 1 & 2: Delete existing emoji (if any) and remove from cache
    await self._delete_discord_emoji(name)

    # Step 3: Find the local file
    local_file_path = _find_emoji_file(self.emojis_path, name)
    if not local_file_path:
        # Error logged in utility function
        logger.error(f"Resync failed for '{name}': Could not find local file.")
        return None

    # Step 4: Process the found emoji file
    new_emoji, _ = await self._process_emoji_file(local_file_path)

    if new_emoji:
        logger.info(f"Resync completed successfully for '{name}'. New ID: {new_emoji.id}")
    else:
        logger.error(f"Resync failed for '{name}' during creation step.")

    logger.info(f"Resync process for emoji '{name}' finished.")  # Log finish regardless of success
    return new_emoji

async def delete_all_emojis(self) -> tuple[list[str], list[str]]:
    """Delete all application emojis that match names from the emoji assets directory.

    This method:
    1. Ensures the emoji cache is initialized
    2. Finds all potential emoji names from the assets directory
    3. Deletes any matching emojis from Discord and updates the cache

    Returns
    -------
    tuple[list[str], list[str]]
        A tuple containing:
        - A list of successfully deleted emoji names
        - A list of emoji names that failed to delete or weren't found
    """
    if not await self._ensure_initialized():
        logger.error("Cannot delete emojis: Cache initialization failed.")
        return [], []

    logger.info("Starting deletion of all application emojis matching asset directory...")

    # Get all potential emoji names from the asset directory
    emoji_names_to_delete: set[str] = set()
    try:
        for file_path in self.emojis_path.iterdir():
            if file_path.is_file() and _is_valid_emoji_name(file_path.stem):
                emoji_names_to_delete.add(file_path.stem)
    except OSError as e:
        logger.error(f"Failed to list files in emoji directory {self.emojis_path}: {e}")
        return [], []

    if not emoji_names_to_delete:
        logger.warning(f"No valid emoji names found in directory: {self.emojis_path}")
        return [], []

    deleted_names: list[str] = []
    failed_names: list[str] = []

    # Process each emoji name
    for emoji_name in emoji_names_to_delete:
        logger.debug(f"Attempting to delete emoji: '{emoji_name}'")

        if await self._delete_discord_emoji(emoji_name):
            deleted_names.append(emoji_name)
        else:
            failed_names.append(emoji_name)

    logger.info(
        f"Emoji deletion finished. Deleted: {len(deleted_names)}, Failed/Not Found: {len(failed_names)}.",
    )

    return deleted_names, failed_names