Utils

active_channels(redis, hours=1, mins=0)

Channel ids with recent message activity.

Parameters:

Name	Type	Description	Default
redis	Redis	Redis connection	required
hours	int	Hours ago to include message history	1
mins	int	Minutes ago to include message history	0

Returns:

Type	Description
list[str]	list[str]: List of channel ids

Source code in services/bot/src/pbot/utils.py

def active_channels(redis: Redis, hours: int=1, mins: int=0) -> list[str]:
    '''Channel ids with recent message activity.

    Args:
        redis (Redis): Redis connection
        hours (int): Hours ago to include message history
        mins (int): Minutes ago to include message history

    Returns:
        list[str]: List of channel ids
    '''
    cutoff = (datetime.now() - timedelta(hours=hours, minutes=mins)).timestamp()

    messages: Any = redis.zrangebyscore(REDIS_MESSAGES_KEY, cutoff, '+inf')

    channels = set()
    for message in messages:
        _, channel_id = (message.split('-')[0]).split('.')
        channels.add(channel_id)
    return list(channels)

channel_message_ids(redis, channel_id, hours=1, mins=0)

Returns a list of message ids from a channel within timeframe.

Parameters:

Name	Type	Description	Default
redis	Redis	Redis connection	required
channel_id	str	Channel to pull messages	required
hours	int	include messages	1
mins	int	include messages	0

Returns:

Name	Type	Description
list	Any	List of message ids (strings)

Source code in services/bot/src/pbot/utils.py

def channel_message_ids(redis: Redis, channel_id: str, hours: int=1, mins: int=0) -> Any:
    '''Returns a list of message ids from a channel within timeframe.

    Args:
        redis (Redis): Redis connection
        channel_id (str): Channel to pull messages
        hours (int): include messages
        mins (int): include messages

    Returns:
        list: List of message ids (strings)
    '''
    msg_key = f'{REDIS_CHANNEL_KEY_PREFIX}:{channel_id}:{REDIS_MESSAGES_KEY}'

    cutoff = (datetime.now() - timedelta(hours=hours, minutes=mins)).timestamp()

    return redis.zrangebyscore(msg_key, cutoff, '+inf')

count_tokens(input, encoding=DEFAULT_TOKEN_ENCODING)

(DEPRECATION WARNING: Will be moved to OpenAi-middleware.) Tokens in a string.

Parameters:

Name	Type	Description	Default
input	str	Sting to count	required
encoding	str	Token encoding	DEFAULT_TOKEN_ENCODING

Returns:

Name	Type	Description
int	int	Count

Source code in services/bot/src/pbot/utils.py

def count_tokens(input: str, encoding: str=DEFAULT_TOKEN_ENCODING) -> int:
    '''(DEPRECATION WARNING: Will be moved to OpenAi-middleware.)
    Tokens in a string.

    Args:
        input (str): Sting to count
        encoding (str): Token encoding

    Returns:
        int: Count
    '''
    encoding = tiktoken.get_encoding(encoding)
    num_tokens = len(encoding.encode(input))
    return num_tokens

create_response(redis, id, content, msg_id)

Creates a response entry in Redis.

Parameters:

Name	Type	Description	Default
redis	Redis	Redis connection	required
id	str	Id for response	required
content	str	Response content	required
msg_id	str	Message id responding to	required

Returns:

Type	Description
None	None

Source code in services/bot/src/pbot/utils.py

def create_response(redis: Redis, id: str, content: str, msg_id: str) -> None:
    '''Creates a response entry in Redis.

    Args:
        redis (Redis): Redis connection
        id (str): Id for response
        content (str): Response content
        msg_id (str): Message id responding to

    Returns:
        None
    '''
    # Get message responding to.
    key_message = get_messages(redis, [msg_id])[0]

    server_id = key_message['origin']['server']['id']
    channel_id = key_message['origin']['server']['channel']['id']
    user_id = key_message['user']['id']

    # Create response entry.
    mapping:dict = Response().mapping()
    mapping['id'] = id
    mapping['user'] = user_id
    mapping['content'] = content
    mapping['message'] = msg_id
    mapping['channel'] = channel_id
    mapping['server'] = server_id
    mapping['time'] = datetime.now().timestamp()

    redis.hset(f'{REDIS_RESPONSE_KEY_PREFIX}:{id}', mapping=mapping)

    # Create links to response entry.
    k = f'{server_id}.{channel_id}.{user_id}-{id}'
    redis.zadd(REDIS_RESPONSES_KEY, {k: datetime.now().timestamp()})
    redis.json().set(f'{REDIS_MESSAGE_KEY_PREFIX}:{msg_id}', '.response', id)

get_messages(redis, ids)

Messages from a list of ids. Ignores unknown ids.

Parameters:

Name	Type	Description	Default
redis	Redis	Redis connection	required
ids	list[str]	(list): List of message ids	required

Returns:

Type	Description
list[dict]	list[dict]: List of message dicts

Source code in services/bot/src/pbot/utils.py

def get_messages(redis: Redis, ids: list[str]) -> list[dict]:
    '''Messages from a list of ids. Ignores unknown ids.

    Args:
        redis (Redis): Redis connection
        ids: (list): List of message ids

    Returns:
        list[dict]: List of message dicts
    '''
    messages = []
    for message_id in ids:
        message = redis.json().get(f'{REDIS_MESSAGE_KEY_PREFIX}:{message_id}')
        if message:
            messages.append(message)

    return messages

is_refusal(content)

(DEPRECATION WARNING: Will be moved to OpenAi-middleware.) Was a AI refusal.

Parameters:

Name	Type	Description	Default
content	str	String	required

Returns:

Name	Type	Description
bool	bool	Boolean

Source code in services/bot/src/pbot/utils.py

def is_refusal(content: str) -> bool:
    '''(DEPRECATION WARNING: Will be moved to OpenAi-middleware.)
    Was a AI refusal.

    Args:
        content (str): String

    Returns:
        bool: Boolean
    '''
    # FIXME: this is jank at best...
    refusal_substrings = [
        'Sorry, but I can\'t',
        'I\'m sorry, but',
        'I cannot comply',
        'As an AI',
        'As a large language model',
        'as a llm',
        'As an AI language model',
        'I apologize, but']

    for substring in refusal_substrings:
        if substring.lower() in content.lower():
            return True

    return False

mark_as_read(redis, messages)

Mark messages are read.

Parameters:

Name	Type	Description	Default
redis	Redis	Redis connection	required
messages	list	List of messages to be marked	required

Returns:

Type	Description
None	None

Source code in services/bot/src/pbot/utils.py

def mark_as_read(redis: Redis, messages: list[dict]) -> None:
    '''Mark messages are read.

    Args:
        redis (Redis): Redis connection
        messages (list): List of messages to be marked

    Returns:
        None
    '''
    for message in messages:
        redis.json().set(
            f'{REDIS_MESSAGE_KEY_PREFIX}:{message["id"]}',
            '.read',
            datetime.now().timestamp())