Discord Message Logging Utility Class Using Message Forwarding Feature (Requires discord.py >= 2.5)

message_forwarder.py

import asyncio
from datetime import datetime
from discord import Member, User, Message, TextChannel, Thread, Embed, Color, HTTPException
from typing import TypedDict


class MessageInfo(TypedDict):
    forwarded_msg: Message
    author: Member | User
    timestamp: datetime
    is_text_only: bool


class MessageForwarder:
    """Utility class for forwarding messages with rate limit handling."""

    BATCH_SIZE = 5
    FORWARD_TIMEOUT = 10
    TEXT_TIMEOUT = 5

    @staticmethod
    def is_text_only(message: Message) -> bool:
        return message.content and not (message.attachments or message.embeds or message.components or message.poll)
    
    @staticmethod
    async def forward_message(target_channel: TextChannel | Thread, message: Message) -> MessageInfo | None:
        """Forward a single message to a target channel.
        
        Args:
            message: The message to forward
            target_channel: The channel to forward the message to

        Returns:
            A tuple containing the author of the original message and the forwarded message, or None if the message could not be forwarded
        """

        try:
            if MessageForwarder.is_text_only(message):
                raise Exception("Message is text only")
            
            # Add timeout to the forward operation
            try:
                forwarding_task = message.forward(target_channel, fail_if_not_exists=True)
                forwarded_message = await asyncio.wait_for(forwarding_task, timeout=MessageForwarder.FORWARD_TIMEOUT)

            except HTTPException as e:
                if e.status == 400:
                    # Force text version for 400 Bad Request errors
                    raise Exception(f"Bad request when forwarding, probably forwarded msg with polls")
                return None
            
            return {
                'forwarded_msg': forwarded_message,
                'author': message.author,
                'timestamp': message.created_at,
                'is_text_only': False
            }
        
        except asyncio.TimeoutError:
            return None
        
        except Exception:
            try:
                text = ""

                embed = Embed(
                    title=f"Deleted Message sent by {message.author.name}",
                    description=message.content or 'Message Component Cannot be Forwarded Normally',
                    timestamp=message.created_at,
                    color=Color.red()
                )
                # Optionally add more fields, e.g. author, channel info, etc.
                embed.add_field(name="Author", value=message.author.mention, inline=True)
                embed.add_field(name="Channel", value=message.channel.mention, inline=True)

                embed.set_author(name=message.author.display_name, icon_url=message.author.avatar.url if message.author.avatar else None)
            
                return {
                    'author': message.author,
                    'forwarded_msg': await target_channel.send(content=text if text else None, embed=embed),
                    'timestamp': message.created_at,
                    'is_text_only': True
                }

            except:
                return None
    
    @staticmethod
    async def forward_bulk_messages(messages: list[Message], target_channel: TextChannel | Thread, specify_author: bool = False) -> None:
        """Forward multiple messages with rate limit handling.
        
        Args:
            messages: List of messages to forward
            target_channel: The channel to forward messages to
            specify_author: Whether to add author information
        """
            
        # Process messages in batches
        message_infos: list[MessageInfo] = []
        i = 0 # Cooldown for normal messages
        j = 0 # Cooldown for forwarding messages
        just_hit_j = False # If we just hit batch for j we timeout only for this time

        while i < len(messages):
            just_hit_j = False
            message_info = await MessageForwarder.forward_message(target_channel, messages[i])
            if message_info:
                message_infos.append(message_info)
                if not message_info['is_text_only']:
                    j += 1
                    if j % MessageForwarder.BATCH_SIZE == 0:
                        just_hit_j = True
            i += 1
            if just_hit_j:
                await asyncio.sleep(MessageForwarder.FORWARD_TIMEOUT)
            elif i % MessageForwarder.BATCH_SIZE == 0:
                await asyncio.sleep(MessageForwarder.TEXT_TIMEOUT)

        # If specify_author is True, add an embed at the end which specifies the author of each message and a link to the forwarded message
        if specify_author and message_infos:
            embed = Embed(title="Authors of Forwarded Messages", color=Color.blue())
            embed.description = '\n'.join(
                [f"- {message_info['author'].mention}: {message_info['forwarded_msg'].jump_url}"
                 for message_info in message_infos if not message_info['is_text_only']]
            )
                    
            await target_channel.send(embed=embed)


__all__ = ['MessageForwarder']