Utils

chunk_str(string, size)

Breaks string on spaces by maximum character length. Useful for breaking responses by character limits.

Parameters:

Name	Type	Description	Default
string	str	String to chunk	required

size (int): Character length to chunk by

Returns:

Type	Description
list[str]	list[str]: A list of strings

Source code in services/transceiver/src/transceiver/utils.py

def chunk_str(string: str, size: int) -> list[str]:
    '''Breaks string on spaces by maximum character length. Useful for breaking
    responses by character limits.


    Args:
		string (str): String to chunk
        size (int): Character length to chunk by

    Returns:
		list[str]: A list of strings
    '''

    if len(string) > size:
        chunks = []
        current_line = ''
        words = string.split(' ')

        for word in words:
            # If str wouldn't exceed chunk size:
            if (len(current_line) + len(word) + 1) < size:
                # If its the initial, empty line.
                if current_line == '':
                    current_line = str(word)
                # Otherwise append + space.
                else:
                    current_line += ' '+str(word)
            # Otherwise, start a new chunk
            else:
                chunks.append(current_line)
                current_line = str(word)

        # Append last chunk segment.
        chunks.append(current_line)

        return chunks

    return [string]

format_message_for_log(message)

(DEPRECATION WARNING: will add to_s() method to Message instead.) String representation of Discord Message.

Parameters:

Name	Type	Description	Default
message	Message	Discord Message	required

Returns:

Name	Type	Description
str	str	Formatted Message object

Source code in services/transceiver/src/transceiver/utils.py

def format_message_for_log(message: discord.Message) -> str:
    '''(DEPRECATION WARNING: will add to_s() method to Message instead.) String
    representation of Discord Message.

    Args:
        message (discord.Message): Discord Message

    Returns:
        str: Formatted Message object
    '''

    # Log statement template.
    tremplate = '{created}|{msg_id}|{server}.{channel}|{author}({nick}):{content}'

    # Populate template.
    tremplate = tremplate.replace('{created}', str(message.created_at))
    tremplate = tremplate.replace('{msg_id}', str(message.id))
    tremplate = tremplate.replace('{server}', message.guild.name)
    tremplate = tremplate.replace('{channel}', message.channel.name)
    tremplate = tremplate.replace('{author}', message.author.name)
    tremplate = tremplate.replace('{nick}', str(message.author.nick))
    tremplate = tremplate.replace('{content}', message.content)

    return tremplate

get_unsent_responses(redis)

(DEPRECATION WARNING: renaming to unsent_responses.) Gets unsent responses waiting in Redis.

Parameters:

Name	Type	Description	Default
redis	Redis	Redis connection	required

Returns:
    list[dict]: A list of unsent response dicts

Source code in services/transceiver/src/transceiver/utils.py

def get_unsent_responses(redis: redis.Redis) -> list[dict]:
    '''(DEPRECATION WARNING: renaming to unsent_responses.) Gets unsent
    responses waiting in Redis.

    Args:
		redis (redis.Redis): Redis connection

	Returns:
	    list[dict]: A list of unsent response dicts
    '''
    cutoff = (datetime.now() - RESPONSE_DELVE_TIME).timestamp()

    # Get recent responses.
    responses: Any = redis.zrangebyscore(
        REDIS_KEY_RESPONSES, cutoff,
        '+inf',
        withscores=False)

    unsent = []

    for mixed_key in responses: # response:<server_id>.<channel_id>.<user_id>-<response_id>

        # Remove prefix
        mixed_key = mixed_key.replace(f'{REDIS_KEY_RESPONSE_PREFIX}:', '') # <server_id>.<channel_id>.<user_id>-<response_id>

        # Unpack ids from mixed key.
        server_channel_user, resp_id = mixed_key.split('-') # "<server_id>.<channel_id>.<user_id>", "<response_id>"
        server_id, channel_id, user_id = server_channel_user.split('.') # ""<server_id>", "<channel_id>", "<user_id>"

        # Pull related objects.
        # TODO: Handle missing...
        response = redis.hgetall(f'{REDIS_KEY_RESPONSE_PREFIX}:{resp_id}')
        server = redis.hgetall(f'server:{server_id}')
        channel = redis.hgetall(f'channel:{channel_id}')
        user = redis.hgetall(f'user:{user_id}')

        # Handle an unsent response.
        if response and (response['sent'] == None or response['sent'] == ''):

            # Ensure Server, Channel, and User are allowed responses.
            if all([
                int(server['respond']) == 1,
                int(channel['respond']) == 1,
                int(user['respond']) == 1
            ]): # FIXME: handle ignored responses clogging queue.
                unsent.append(response)

    return unsent

handle_message(redis, message)

Process an incoming Discord message.

Parameters:

Name	Type	Description	Default
redis	Redis	Redis connection	required
message	Message	Discord Message	required

Returns:

Type	Description
None	None

Source code in services/transceiver/src/transceiver/utils.py

def handle_message(redis: redis.Redis, message: discord.Message) -> None:
    '''Process an incoming Discord message.

    Args:
        redis (redis.Redis): Redis connection
        message (discord.Message): Discord Message

    Returns:
        None
    '''

    logger.debug(format_message_for_log(message))
    process_msg(redis, message)

handle_response(discord_client, redis_conn, response) async

Asynchronously send a single/multi-part message to Discord.

Parameters:

Name	Type	Description	Default
response	dict[str, id]	Dictionary represensation	required

Returns:

Type	Description
None	None

Source code in services/transceiver/src/transceiver/utils.py

async def handle_response(discord_client: discord.Client, redis_conn: redis.Redis, response: dict[str, any]) -> None:
    '''Asynchronously send a single/multi-part message to Discord.

    Args:
        response (dict[str, id]): Dictionary represensation

    Returns:
        None
    '''

    # Find Discord message to respond to via API calls.
    channel: Any = discord_client.get_channel(int(response['channel']))
    message: Any = channel.get_partial_message(int(response['message']))

    # Handle multi-part responses due to length.
    if len(response['content']) > DISCORD_CHARACTER_LIMIT:
        chunks = chunk_str(response['content'], DISCORD_CHARACTER_LIMIT)

        for chunk in chunks:
            # Await each message so the response messages are in order.
            await send_message(message, chunk)

    else:
       await send_message(message, response["content"])

    # Mark response as sent.
    redis_conn.hset(
        f'{REDIS_KEY_RESPONSE_PREFIX}:{response["id"]}',
        'sent', str(datetime.now().timestamp())
    )

intents()

Returns a configured Intents object for a Discord a client.

By default, a bot does not have access to messages or their contents. Permission has to be explicitly given.

See docs: https://discordpy.readthedocs.io/en/stable/intents.html

Returns:

Type	Description
Intents	discord.Intents: Discord Intents object for client.

Source code in services/transceiver/src/transceiver/utils.py

def intents() -> discord.Intents:
	'''Returns a configured Intents object for a Discord a client.

    By default, a bot does not have access to messages or their contents.
    Permission has to be explicitly given.

    See docs: https://discordpy.readthedocs.io/en/stable/intents.html

    Returns:
        discord.Intents: Discord Intents object for client.
	'''

	intents = discord.Intents.default()
	intents.messages = True
	intents.message_content = True
	return intents

send_message(message, content) async

Asynchronously sends a response to a specific Discord message.

Parameters:

Name	Type	Description	Default
message	Message	Discord message to respond	required
content	str	Response body	required

Returns:

Type	Description
None	None

Source code in services/transceiver/src/transceiver/utils.py

async def send_message(message: discord.Message, content: str) -> None:
    '''Asynchronously sends a response to a specific Discord message.

    Args:
        message (discord.Message): Discord message to respond
        content (str): Response body

    Returns:
        None
    '''

    try:
        await message.reply(content)
    except Exception as error: # FIXME: too permissive--don't try forever
        logger.error(error)