Actions

Common message actions

Actions commonly run on chat messages

This file is part of Rumble Chat Actor.

Rumble Chat Actor is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version.

Rumble Chat Actor is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.

You should have received a copy of the GNU General Public License along with Rumble Chat Actor. If not, see https://www.gnu.org/licenses/.

S.D.G

ChatBlipper

Blip with chat activity, getting fainter as activity gets more common

Source code in rumchat_actor/actions.py

class ChatBlipper:
    """Blip with chat activity, getting fainter as activity gets more common"""

    def __init__(self, sound_filename: str, rarity_regen_time=60, stay_dead_time=10, rarity_reduce=0.1):
        """Blip with chat activity, getting fainter as activity gets more common.
    Instance this object, then pass it to RumbleChatActor().register_message_action()

    Args:
        sound_filename (str): The filename of the blip sound to play.
        rarity_regen_time (int): How long before the blip volume regenerates to maximum, in seconds.
        stay_dead_time (int): Effectively more regen time, in seconds, but with the volume staying at zero for the duration.
        rarity_reduce (float): How much a message reduces the volume in factor, ranging from >0 to 1."""

        self.sound = None
        self.load_sound(sound_filename)
        self.rarity_regen_time = rarity_regen_time
        self.stay_dead_time = stay_dead_time
        self.rarity_reduce = rarity_reduce

        # Time in the past at which we would have been silent
        self.silent_time = 0

    def load_sound(self, fn):
        """Load a sound from a file to use as a blip"""
        # Make sure PyGame mixer is initialized
        if not mixer.get_init():
            mixer.init()

        self.sound = mixer.Sound(fn)

    @property
    def current_volume(self):
        """Calculate the current volume based on how rare messages have been, from 0 to 1"""
        return max((min((time.time() - self.silent_time) / self.rarity_regen_time, 1), 0))

    def reduce_rarity(self):
        """Reduce the remembered rarity of a message"""
        curtime = time.time()

        # Limit the effective regen to 100%
        if curtime - self.silent_time > self.rarity_regen_time:
            self.silent_time = curtime - self.rarity_regen_time

        # Move the time we "were" silent forward, capping at present + stay-dead time
        self.silent_time = min((self.silent_time + self.rarity_regen_time *
                               self.rarity_reduce, curtime + self.stay_dead_time))

    def action(self, message, act_props, actor):
        """Blip for a chat message, taking rarity into account for the volume

    Args:
        message (cocorum.chatapi.Message): The chat message to run this action on.
        act_props (dict): Action properties, aka metadata about what other things did with this message
        actor (RumbleChatActor): The chat actor.

    Returns:
        act_props (dict): Dictionary of additional recorded properties from running this action."""

        self.sound.set_volume(self.current_volume)
        self.sound.play()
        self.reduce_rarity()
        return {}

current_volume property

Calculate the current volume based on how rare messages have been, from 0 to 1

__init__(sound_filename, rarity_regen_time=60, stay_dead_time=10, rarity_reduce=0.1)

Blip with chat activity, getting fainter as activity gets more common. Instance this object, then pass it to RumbleChatActor().register_message_action()

Parameters:

Name	Type	Description	Default
sound_filename	str	The filename of the blip sound to play.	required
rarity_regen_time	int	How long before the blip volume regenerates to maximum, in seconds.	60
stay_dead_time	int	Effectively more regen time, in seconds, but with the volume staying at zero for the duration.	10
rarity_reduce	float	How much a message reduces the volume in factor, ranging from >0 to 1.	0.1

Source code in rumchat_actor/actions.py

def __init__(self, sound_filename: str, rarity_regen_time=60, stay_dead_time=10, rarity_reduce=0.1):
    """Blip with chat activity, getting fainter as activity gets more common.
Instance this object, then pass it to RumbleChatActor().register_message_action()

Args:
    sound_filename (str): The filename of the blip sound to play.
    rarity_regen_time (int): How long before the blip volume regenerates to maximum, in seconds.
    stay_dead_time (int): Effectively more regen time, in seconds, but with the volume staying at zero for the duration.
    rarity_reduce (float): How much a message reduces the volume in factor, ranging from >0 to 1."""

    self.sound = None
    self.load_sound(sound_filename)
    self.rarity_regen_time = rarity_regen_time
    self.stay_dead_time = stay_dead_time
    self.rarity_reduce = rarity_reduce

    # Time in the past at which we would have been silent
    self.silent_time = 0

action(message, act_props, actor)

Blip for a chat message, taking rarity into account for the volume

Parameters:

Name	Type	Description	Default
message	Message	The chat message to run this action on.	required
act_props	dict	Action properties, aka metadata about what other things did with this message	required
actor	RumbleChatActor	The chat actor.	required

Returns:

Name	Type	Description
act_props	dict	Dictionary of additional recorded properties from running this action.

Source code in rumchat_actor/actions.py

def action(self, message, act_props, actor):
    """Blip for a chat message, taking rarity into account for the volume

Args:
    message (cocorum.chatapi.Message): The chat message to run this action on.
    act_props (dict): Action properties, aka metadata about what other things did with this message
    actor (RumbleChatActor): The chat actor.

Returns:
    act_props (dict): Dictionary of additional recorded properties from running this action."""

    self.sound.set_volume(self.current_volume)
    self.sound.play()
    self.reduce_rarity()
    return {}

load_sound(fn)

Load a sound from a file to use as a blip

Source code in rumchat_actor/actions.py

def load_sound(self, fn):
    """Load a sound from a file to use as a blip"""
    # Make sure PyGame mixer is initialized
    if not mixer.get_init():
        mixer.init()

    self.sound = mixer.Sound(fn)

reduce_rarity()

Reduce the remembered rarity of a message

Source code in rumchat_actor/actions.py

def reduce_rarity(self):
    """Reduce the remembered rarity of a message"""
    curtime = time.time()

    # Limit the effective regen to 100%
    if curtime - self.silent_time > self.rarity_regen_time:
        self.silent_time = curtime - self.rarity_regen_time

    # Move the time we "were" silent forward, capping at present + stay-dead time
    self.silent_time = min((self.silent_time + self.rarity_regen_time *
                           self.rarity_reduce, curtime + self.stay_dead_time))

RantTTSManager

System to TTS rant messages, with threshhold settings

Source code in rumchat_actor/actions.py

class RantTTSManager():
    """System to TTS rant messages, with threshhold settings"""

    def __init__(self):
        """System to TTS rant messages, with threshhold settings.
    Instance this object, then pass it to RumbleChatActor().register_message_action()"""

        # The amount a rant must be to be TTS-ed
        self.__tts_amount_threshold = 0

        # The TTS callable to use
        self.__say = talkey.Talkey().say

    @property
    def tts_amount_threshold(self):
        """The amount a rant must be to be TTS-ed, 0 means all rants are TTS-ed"""
        return self.__tts_amount_threshold

    @tts_amount_threshold.setter
    def tts_amount_threshold(self, new):
        """The amount a rant must be to be TTS-ed, 0 means all rants are TTS-ed

    Args:
        new (int): The new threshold in cents."""

        assert isinstance(
            new, (int, float)) and new >= 0, "Value must be a number greater than zero"
        self.__tts_amount_threshold = new

    def set_rant_tts_sayer(self, new):
        """Set the callable to be used on rant TTS

    Args:
        new (callable): The function or method to call, passing the message text."""

        assert callable(new), "Must be a callable"
        self.__say = new

    def action(self, message, act_props, actor):
        """TTS rants above the manager instance's threshhold

    Args:
        message (cocorum.chatapi.Message): The chat message to run this action on.
        act_props (dict): Action properties, aka metadata about what other things did with this message
        actor (RumbleChatActor): The chat actor.

    Returns:
        act_props (dict): Dictionary of additional recorded properties from running this action."""

        # Do not overlap sounds
        if act_props["sound"]:
            return {}

        if message.is_rant and message.rant_price_cents >= self.__tts_amount_threshold:
            self.__say(message.text)
            return {"sound": True}
        return {}

tts_amount_threshold property writable

The amount a rant must be to be TTS-ed, 0 means all rants are TTS-ed

__init__()

System to TTS rant messages, with threshhold settings. Instance this object, then pass it to RumbleChatActor().register_message_action()

Source code in rumchat_actor/actions.py

def __init__(self):
    """System to TTS rant messages, with threshhold settings.
Instance this object, then pass it to RumbleChatActor().register_message_action()"""

    # The amount a rant must be to be TTS-ed
    self.__tts_amount_threshold = 0

    # The TTS callable to use
    self.__say = talkey.Talkey().say

action(message, act_props, actor)

TTS rants above the manager instance's threshhold

Parameters:

Name	Type	Description	Default
message	Message	The chat message to run this action on.	required
act_props	dict	Action properties, aka metadata about what other things did with this message	required
actor	RumbleChatActor	The chat actor.	required

Returns:

Name	Type	Description
act_props	dict	Dictionary of additional recorded properties from running this action.

Source code in rumchat_actor/actions.py

def action(self, message, act_props, actor):
    """TTS rants above the manager instance's threshhold

Args:
    message (cocorum.chatapi.Message): The chat message to run this action on.
    act_props (dict): Action properties, aka metadata about what other things did with this message
    actor (RumbleChatActor): The chat actor.

Returns:
    act_props (dict): Dictionary of additional recorded properties from running this action."""

    # Do not overlap sounds
    if act_props["sound"]:
        return {}

    if message.is_rant and message.rant_price_cents >= self.__tts_amount_threshold:
        self.__say(message.text)
        return {"sound": True}
    return {}

set_rant_tts_sayer(new)

Set the callable to be used on rant TTS

Parameters:

Name	Type	Description	Default
new	callable	The function or method to call, passing the message text.	required

Source code in rumchat_actor/actions.py

def set_rant_tts_sayer(self, new):
    """Set the callable to be used on rant TTS

Args:
    new (callable): The function or method to call, passing the message text."""

    assert callable(new), "Must be a callable"
    self.__say = new

Thanker

Bases: Thread

Thank followers and subscribers in the chat

Source code in rumchat_actor/actions.py

class Thanker(threading.Thread):
    """Thank followers and subscribers in the chat"""

    def __init__(self, actor, **kwargs):
        """Thank followers and subscribers in the chat.
    Instance this object, then pass it to RumbleChatActor().register_message_action()

    Args:
        actor (RumbleChatActor): The Rumble Chat Actor instance.
        follower_message (str): Message to format with Cocorum Follower object.
            Defaults to static.Thank.DefaultMessages.follower
        subscriber_message (str): Message to format with the Cocorum Subscriber object.
            Defaults to static.Thank.DefaultMessages.subscriber
        gifted_subs_message (str): Message to format with the Cocorum GiftPurchaseNotification object.
            Defaults to static.Thank.DefaultMessages.gifted_subs"""

        super().__init__(daemon=True)
        self.actor = actor
        self.rum_api = self.actor.rum_api
        assert self.rum_api, "Thanker cannot function if actor does not have Rumble API"

        # Set up default messages
        self.follower_message = kwargs.get(
            "follower_message", static.Thank.DefaultMessages.follower)
        self.subscriber_message = kwargs.get(
            "subscriber_message", static.Thank.DefaultMessages.subscriber)
        self.gifted_subs_message = kwargs.get(
            "gifted_subs_message", static.Thank.DefaultMessages.gifted_subs)

        # Start the thread immediately
        self.start()

    def action(self, message, act_props, actor):
        """Check for subscription gifts, and thank for them

    Args:
        message (cocorum.chatapi.Message): The chat message to run this action on.
        act_props (dict): Action properties, aka metadata about what other things did with this message
        actor (RumbleChatActor): The chat actor.

    Returns:
        act_props (dict): Dictionary of additional recorded properties from running this action."""

        gift = message.gift_purchase_notification

        # This is not a gift purchase notification
        if not gift:
            return

        self.actor.send_message(self.gifted_subs_message.format(gift=gift))

        return {}

    def run(self):
        """Continuously check for new followers and subscribers"""
        while self.actor.keep_running:
            # Thank all the new followers
            for follower in self.rum_api.new_followers:
                self.actor.send_message(
                    self.follower_message.format(follower=follower))

            # Thank all the new subscribers
            for subscriber in self.rum_api.new_subscribers:
                self.actor.send_message(
                    self.follower_message.format(subscriber=subscriber))

            # Wait a bit, either the Rumble API refresh rate or the message sending cooldown
            time.sleep(
                max((self.rum_api.refresh_rate, static.Message.send_cooldown)))

__init__(actor, **kwargs)

Thank followers and subscribers in the chat. Instance this object, then pass it to RumbleChatActor().register_message_action()

Parameters:

Name	Type	Description	Default
actor	RumbleChatActor	The Rumble Chat Actor instance.	required
follower_message	str	Message to format with Cocorum Follower object. Defaults to static.Thank.DefaultMessages.follower	required
subscriber_message	str	Message to format with the Cocorum Subscriber object. Defaults to static.Thank.DefaultMessages.subscriber	required
gifted_subs_message	str	Message to format with the Cocorum GiftPurchaseNotification object. Defaults to static.Thank.DefaultMessages.gifted_subs	required

Source code in rumchat_actor/actions.py

def __init__(self, actor, **kwargs):
    """Thank followers and subscribers in the chat.
Instance this object, then pass it to RumbleChatActor().register_message_action()

Args:
    actor (RumbleChatActor): The Rumble Chat Actor instance.
    follower_message (str): Message to format with Cocorum Follower object.
        Defaults to static.Thank.DefaultMessages.follower
    subscriber_message (str): Message to format with the Cocorum Subscriber object.
        Defaults to static.Thank.DefaultMessages.subscriber
    gifted_subs_message (str): Message to format with the Cocorum GiftPurchaseNotification object.
        Defaults to static.Thank.DefaultMessages.gifted_subs"""

    super().__init__(daemon=True)
    self.actor = actor
    self.rum_api = self.actor.rum_api
    assert self.rum_api, "Thanker cannot function if actor does not have Rumble API"

    # Set up default messages
    self.follower_message = kwargs.get(
        "follower_message", static.Thank.DefaultMessages.follower)
    self.subscriber_message = kwargs.get(
        "subscriber_message", static.Thank.DefaultMessages.subscriber)
    self.gifted_subs_message = kwargs.get(
        "gifted_subs_message", static.Thank.DefaultMessages.gifted_subs)

    # Start the thread immediately
    self.start()

action(message, act_props, actor)

Check for subscription gifts, and thank for them

Parameters:

Name	Type	Description	Default
message	Message	The chat message to run this action on.	required
act_props	dict	Action properties, aka metadata about what other things did with this message	required
actor	RumbleChatActor	The chat actor.	required

Returns:

Name	Type	Description
act_props	dict	Dictionary of additional recorded properties from running this action.

Source code in rumchat_actor/actions.py

def action(self, message, act_props, actor):
    """Check for subscription gifts, and thank for them

Args:
    message (cocorum.chatapi.Message): The chat message to run this action on.
    act_props (dict): Action properties, aka metadata about what other things did with this message
    actor (RumbleChatActor): The chat actor.

Returns:
    act_props (dict): Dictionary of additional recorded properties from running this action."""

    gift = message.gift_purchase_notification

    # This is not a gift purchase notification
    if not gift:
        return

    self.actor.send_message(self.gifted_subs_message.format(gift=gift))

    return {}

run()

Continuously check for new followers and subscribers

Source code in rumchat_actor/actions.py

def run(self):
    """Continuously check for new followers and subscribers"""
    while self.actor.keep_running:
        # Thank all the new followers
        for follower in self.rum_api.new_followers:
            self.actor.send_message(
                self.follower_message.format(follower=follower))

        # Thank all the new subscribers
        for subscriber in self.rum_api.new_subscribers:
            self.actor.send_message(
                self.follower_message.format(subscriber=subscriber))

        # Wait a bit, either the Rumble API refresh rate or the message sending cooldown
        time.sleep(
            max((self.rum_api.refresh_rate, static.Message.send_cooldown)))

TimedMessagesManager

System to send messages on a timed basis

Source code in rumchat_actor/actions.py

class TimedMessagesManager():
    """System to send messages on a timed basis"""

    def __init__(self, actor, messages: iter, delay=60, in_between=0):
        """System to send messages on a timed basis. Instance this object, then pass it to RumbleChatActor().register_message_action()

    Args:
        actor (RumbleChatActor): The actor, to send the timed messages,
        messages (list): List of str messages to send
        delay (int): Time between messages in seconds
        in_between (int): Number of messages that must be sent before we send another timed one"""

        self.actor = actor
        assert len(messages) > 0, "List of messages to send cannot be empty"
        self.messages = messages
        assert delay > static.Message.send_cooldown, "Cannot send timed messages that frequently"
        self.delay = delay
        self.in_between = in_between

        # Next message to send
        self.up_next_index = 0

        # Time of last send
        self.last_send_time = 0

        # Counter for messages sent since our last announcement
        self.in_between_counter = 0

        # Start the sender loop thread
        self.running = True
        self.sender_thread = threading.Thread(
            target=self.sender_loop, daemon=True)
        self.sender_thread.start()

    def action(self, message, act_props, actor):
        """Count the messages sent

    Args:
        message (cocorum.chatapi.Message): The chat message to run this action on.
        act_props (dict): Action properties, aka metadata about what other things did with this message
        actor (RumbleChatActor): The chat actor.

    Returns:
        act_props (dict): Dictionary of additional recorded properties from running this action."""

        self.in_between_counter += 1
        return {}

    def sender_loop(self):
        """Continuously wait till it is time to send another message"""
        while self.running:
            # time to send a message?
            if self.in_between_counter >= self.in_between and time.time() - self.last_send_time >= self.delay:
                # Send a message
                self.actor.send_message(self.messages[self.up_next_index])

                # Up the index of the next message, with wrapping
                self.up_next_index += 1
                # self.messages should theoretically never change, so this is thread-safe
                if self.up_next_index >= len(self.messages):
                    self.up_next_index = 0

                # Reset wait counters
                self.in_between_counter = 0
                self.last_send_time = time.time()

            time.sleep(1)

__init__(actor, messages, delay=60, in_between=0)

System to send messages on a timed basis. Instance this object, then pass it to RumbleChatActor().register_message_action()

Parameters:

Name	Type	Description	Default
actor	RumbleChatActor	The actor, to send the timed messages,	required
messages	list	List of str messages to send	required
delay	int	Time between messages in seconds	60
in_between	int	Number of messages that must be sent before we send another timed one	0

Source code in rumchat_actor/actions.py

def __init__(self, actor, messages: iter, delay=60, in_between=0):
    """System to send messages on a timed basis. Instance this object, then pass it to RumbleChatActor().register_message_action()

Args:
    actor (RumbleChatActor): The actor, to send the timed messages,
    messages (list): List of str messages to send
    delay (int): Time between messages in seconds
    in_between (int): Number of messages that must be sent before we send another timed one"""

    self.actor = actor
    assert len(messages) > 0, "List of messages to send cannot be empty"
    self.messages = messages
    assert delay > static.Message.send_cooldown, "Cannot send timed messages that frequently"
    self.delay = delay
    self.in_between = in_between

    # Next message to send
    self.up_next_index = 0

    # Time of last send
    self.last_send_time = 0

    # Counter for messages sent since our last announcement
    self.in_between_counter = 0

    # Start the sender loop thread
    self.running = True
    self.sender_thread = threading.Thread(
        target=self.sender_loop, daemon=True)
    self.sender_thread.start()

action(message, act_props, actor)

Count the messages sent

Parameters:

Name	Type	Description	Default
message	Message	The chat message to run this action on.	required
act_props	dict	Action properties, aka metadata about what other things did with this message	required
actor	RumbleChatActor	The chat actor.	required

Returns:

Name	Type	Description
act_props	dict	Dictionary of additional recorded properties from running this action.

Source code in rumchat_actor/actions.py

def action(self, message, act_props, actor):
    """Count the messages sent

Args:
    message (cocorum.chatapi.Message): The chat message to run this action on.
    act_props (dict): Action properties, aka metadata about what other things did with this message
    actor (RumbleChatActor): The chat actor.

Returns:
    act_props (dict): Dictionary of additional recorded properties from running this action."""

    self.in_between_counter += 1
    return {}

sender_loop()

Continuously wait till it is time to send another message

Source code in rumchat_actor/actions.py

def sender_loop(self):
    """Continuously wait till it is time to send another message"""
    while self.running:
        # time to send a message?
        if self.in_between_counter >= self.in_between and time.time() - self.last_send_time >= self.delay:
            # Send a message
            self.actor.send_message(self.messages[self.up_next_index])

            # Up the index of the next message, with wrapping
            self.up_next_index += 1
            # self.messages should theoretically never change, so this is thread-safe
            if self.up_next_index >= len(self.messages):
                self.up_next_index = 0

            # Reset wait counters
            self.in_between_counter = 0
            self.last_send_time = time.time()

        time.sleep(1)

UserAnnouncer

Announce new users as they arrive in the chat

Source code in rumchat_actor/actions.py

class UserAnnouncer:
    """Announce new users as they arrive in the chat"""

    def __init__(self, announcer: callable = None, known_users=[], special_announcers={}):
        """Announce new users as they arrive in the chat

        Args:
            announcer (callable): Function or method to call. Will be passed
                the message action special of message, act_props, and actor.
                Should return act_props dict likewise.
                Defaults to None.
            known_users (list): List of users not to announce. Useful if you
                only want this announcer to go off for brand new chatters. You
                can save this list after actor exit from this object's
                `.known_users` attribute.
            special_announcers (dict[str, callable]): Dict of username to
                announcer callable pairs. Should take same args as regular
                announcers, but they will only be called on the username that
                matches them."""

        self.announcer = announcer
        self.known_users = known_users
        self.special_announcers = special_announcers

    def action(self, message, act_props, actor):
        """Announce the user if they are new

    Args:
        message (cocorum.chatapi.Message): The chat message to run this action on.
        act_props (dict): Action properties, aka metadata about what other things did with this message
        actor (RumbleChatActor): The chat actor.

    Returns:
        act_props (dict): Dictionary of additional recorded properties from running this action."""

        # We already announced this user
        if message.user.username in self.known_users:
            return

        # Remember this user
        self.known_users.append(message.user.username)

        # We might have a special announcer for this user
        ann = self.special_announcers.get(
            message.user.username, self.announcer)

        # Our main announcer may be None
        if not ann:
            return

        props = ann(message, act_props, actor)
        props["announced"] = True
        return props

__init__(announcer=None, known_users=[], special_announcers={})

Announce new users as they arrive in the chat

Parameters:

Name	Type	Description	Default
announcer	callable	Function or method to call. Will be passed the message action special of message, act_props, and actor. Should return act_props dict likewise. Defaults to None.	None
known_users	list	List of users not to announce. Useful if you only want this announcer to go off for brand new chatters. You can save this list after actor exit from this object's .known_users attribute.	[]
special_announcers	dict[str, callable]	Dict of username to announcer callable pairs. Should take same args as regular announcers, but they will only be called on the username that matches them.	{}

Source code in rumchat_actor/actions.py

def __init__(self, announcer: callable = None, known_users=[], special_announcers={}):
    """Announce new users as they arrive in the chat

    Args:
        announcer (callable): Function or method to call. Will be passed
            the message action special of message, act_props, and actor.
            Should return act_props dict likewise.
            Defaults to None.
        known_users (list): List of users not to announce. Useful if you
            only want this announcer to go off for brand new chatters. You
            can save this list after actor exit from this object's
            `.known_users` attribute.
        special_announcers (dict[str, callable]): Dict of username to
            announcer callable pairs. Should take same args as regular
            announcers, but they will only be called on the username that
            matches them."""

    self.announcer = announcer
    self.known_users = known_users
    self.special_announcers = special_announcers

action(message, act_props, actor)

Announce the user if they are new

Parameters:

Name	Type	Description	Default
message	Message	The chat message to run this action on.	required
act_props	dict	Action properties, aka metadata about what other things did with this message	required
actor	RumbleChatActor	The chat actor.	required

Returns:

Name	Type	Description
act_props	dict	Dictionary of additional recorded properties from running this action.

Source code in rumchat_actor/actions.py

def action(self, message, act_props, actor):
    """Announce the user if they are new

Args:
    message (cocorum.chatapi.Message): The chat message to run this action on.
    act_props (dict): Action properties, aka metadata about what other things did with this message
    actor (RumbleChatActor): The chat actor.

Returns:
    act_props (dict): Dictionary of additional recorded properties from running this action."""

    # We already announced this user
    if message.user.username in self.known_users:
        return

    # Remember this user
    self.known_users.append(message.user.username)

    # We might have a special announcer for this user
    ann = self.special_announcers.get(
        message.user.username, self.announcer)

    # Our main announcer may be None
    if not ann:
        return

    props = ann(message, act_props, actor)
    props["announced"] = True
    return props

ollama_message_moderate(message, act_props, actor)

Moderate a message with Ollama, deleting if needed

Parameters:

Name	Type	Description	Default
message	Message	The chat message to run this action on.	required
act_props	dict	Action properties, aka metadata about what other things did with this message	required
actor	RumbleChatActor	The chat actor.	required

Returns:

Name	Type	Description
act_props	dict	Dictionary of recorded properties from running this action.

Source code in rumchat_actor/actions.py

def ollama_message_moderate(message, act_props, actor):
    """Moderate a message with Ollama, deleting if needed

    Args:
        message (cocorum.chatapi.Message): The chat message to run this action on.
        act_props (dict): Action properties, aka metadata about what other things did with this message
        actor (RumbleChatActor): The chat actor.

    Returns:
        act_props (dict): Dictionary of recorded properties from running this action."""

    assert OLLAMA_IMPORTED, "The Ollama library and a working Ollama installation are required for ollama_message_moderate"

    # Message was blank
    if not message.text.strip():
        print("Message was blank.")
        return {}

    # User has an immunity badge
    if True in [badge in message.user.badges for badge in static.Moderation.staff_badges]:
        print(f"{message.user.username} is staff, skipping LLM check.")
        return {}

    # Get the LLM verdict
    response = ollama.chat(model=static.AutoModerator.llm_model, messages=[
        {"role": "system", "content": static.AutoModerator.llm_sys_prompt},
        {"role": "user", "content": message.text},
    ])

    # Parse the verdict
    try:
        verdict = int(response["message"]["content"])

    # Verdict was not valid
    except ValueError:
        print(
            f"Bad verdict for {message.text} : {response["message"]["content"]}")
        return {}

    # Response was not in expected format
    except KeyError:
        print(
            f"Could not get verdict for {message.text} : Response: {response}")
        return {}

    # Returned 1 for SFW
    if verdict:
        print("LLM verdicted as clean: " + message.text)
        return {}

    # Returned 0 for NSFW
    print("LLM verdicted as dirty: " + message.text)
    actor.delete_message(message)
    return {"deleted": True}

This file is part of Rumble Chat Actor.

Rumble Chat Actor is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version.

Rumble Chat Actor is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.

You should have received a copy of the GNU General Public License along with Rumble Chat Actor. If not, see https://www.gnu.org/licenses/.

S.D.G.