cocorum.chatapi

The primary use from this module is the ChatAPI class, a wrapper for Rumble's internal chat API. When passed credentials, it currently automatically creates an internal instance of cocorum.servicephp.ServicePHP() to log in with, and includes links to the chat-related methods of ServicePHP() (muting users for example). All other classes are supporting sub-classes.

Internal chat API client

This part of cocorum is not part of the official Rumble Live Stream API, but may provide a more reliable method of ensuring all chat messages are received. It also can do to-chat interactions, sometimes via Service.PHP.

Example usage:

from cocorum import chatapi

#Additionally pass username and password for to-chat interactions
chat = chatapi.ChatAPI(stream_id = STREAM_ID) #Stream ID can be base 10 or 36
chat.clear_mailbox() #Erase messages that were still visible before we connected

#Get messages for one minute
start_time = time.time()
while time.time() - start_time < 60 and (msg := chat.get_message()):
    print(msg.user.username, "said", msg)

S.D.G.

ChatAPI

The Rumble internal chat API

Source code in cocorum/chatapi.py

class ChatAPI():
    """The Rumble internal chat API"""
    def __init__(self, stream_id, username: str = None, password: str = None, session = None, history_len = 1000):
        """The Rumble internal chat API

    Args:
        stream_id (int, str): Stream ID in base 10 or 36.
        username (str): Username to login with.
            Defaults to no login.
        password (str): Password to log in with.
            Defaults to no login.
        session (str, dict): Session token or cookie dict to authenticate with.
            Defaults to getting new session with username and password.
        history_len (int): Length of message history to store.
            Defaults to 1000.
            """

        self.stream_id = utils.ensure_b36(stream_id)

        self.__mailbox = [] #A mailbox if you will
        self.__history = [] #Chat history
        self.history_len = history_len #How many messages to store in history
        self.pinned_message = None #If a message is pinned, it is assigned to this
        self.users = {} #Dictionary of users by user ID
        self.channels = {} #Dictionary of channels by channel ID
        self.badges = {}

        #Generate our URLs
        self.sse_url = static.URI.ChatAPI.sse_stream.format(stream_id_b10 = self.stream_id_b10)
        self.message_api_url = static.URI.ChatAPI.message.format(stream_id_b10 = self.stream_id_b10)

        #Connect to SSE stream
        #Note: We do NOT want this request to have a timeout
        response = requests.get(self.sse_url, stream = True, headers = static.RequestHeaders.sse_api)
        self.client = sseclient.SSEClient(response)
        self.event_generator = self.client.events()
        self.chat_running = True
        self.parse_init_data(self.__next_event_json())

        #If we have session login, use them
        if (username and password) or session:
            self.servicephp = ServicePHP(username, password, session)
            self.scraper = scraping.Scraper(self.servicephp)
        else:
            self.servicephp = None

        #The last time we sent a message
        self.last_send_time = 0

    @property
    def session_cookie(self):
        """The session cookie we are logged in with"""
        if self.servicephp:
            return self.servicephp.session_cookie
        return None

    @property
    def history(self):
        """The chat history, trimmed to history_len"""
        return tuple(self.__history)

    def send_message(self, text: str, channel_id: int = None):
        """Send a message in chat.

    Args:
        text (str): The message text.
        channel_id (int): Numeric ID of the channel to use.
            Defaults to None.

    Returns:
        ID (int): The ID of the sent message.
        User (ChatAPIUser): Your current chat user information.
        """

        assert self.session_cookie, "Not logged in, cannot send message"
        assert len(text) <= static.Message.max_len, "Mesage is too long"
        curtime = time.time()
        assert self.last_send_time + static.Message.send_cooldown <= curtime, "Sending messages too fast"
        assert utils.options_check(self.message_api_url, "POST"), "Rumble denied options request to post message"
        r = requests.post(
            self.message_api_url,
            cookies = self.session_cookie,
            json = {
                "data": {
                    "request_id": utils.generate_request_id(),
                    "message": {
                        "text": text
                    },
                    "rant": None,
                    "channel_id": channel_id
                    }
                },
            # headers = static.RequestHeaders.user_agent,
            timeout = static.Delays.request_timeout,
            )

        if r.status_code != 200:
            print("Error: Sending message failed,", r, r.text)
            return

        return int(r.json()["data"]["id"]), ChatAPIUser(r.json()["data"]["user"], self)

    def command(self, command_message: str):
        """Send a native chat command

    Args:
        command_message (str): The message you would send to launch this command in chat.

    Returns:
        JSON (dict): The JSON returned by the command.
        """

        assert command_message.startswith(static.Message.command_prefix), "Not a command message"
        r = requests.post(
            static.URI.ChatAPI.command,
            data = {
                "video_id" : self.stream_id_b10,
                "message" : command_message,
                },
            cookies = self.session_cookie,
            headers = static.RequestHeaders.user_agent,
            timeout = static.Delays.request_timeout,
            )
        assert r.status_code == 200, f"Command failed: {r}\n{r.text}"
        return r.json()

    def delete_message(self, message):
        """Delete a message in chat.

    Args:
        message (int): Object which when converted to integer is the target message ID.
        """

        assert self.session_cookie, "Not logged in, cannot delete message"
        assert utils.options_check(self.message_api_url + f"/{int(message)}", "DELETE"), "Rumble denied options request to delete message"

        r = requests.delete(
            self.message_api_url + f"/{int(message)}",
            cookies = self.session_cookie,
            # headers = static.RequestHeaders.user_agent,
            timeout = static.Delays.request_timeout,
            )

        if r.status_code != 200:
            print("Error: Deleting message failed,", r, r.content.decode(static.Misc.text_encoding))
            return False

        return True

    def pin_message(self, message):
        """Pin a message"""
        assert self.session_cookie, "Not logged in, cannot pin message"
        return self.servicephp.chat_pin(self.stream_id_b10, message)

    def unpin_message(self, message = None):
        """Unpin the pinned message"""
        assert self.session_cookie, "Not logged in, cannot unpin message"
        if not message:
            message = self.pinned_message
        assert message, "No known pinned message and ID not provided"
        return self.servicephp.chat_pin(self.stream_id_b10, message, unpin = True)

    def mute_user(self, user, duration: int = None, total: bool = False):
        """Mute a user.

    Args:
        user (str): Username to mute.
        duration (int): How long to mute the user in seconds.
            Defaults to infinite.
        total (bool): Wether or not they are muted across all videos.
            Defaults to False, just this video.
            """

        assert self.session_cookie, "Not logged in, cannot mute user"
        return self.servicephp.mute_user(
            username = str(user),
            is_channel = False,
            video = self.stream_id_b10,
            duration = duration,
            total = total
            )

    def unmute_user(self, user):
        """Unmute a user.

    Args:
        user (str): Username to unmute
        """

        assert self.session_cookie, "Not logged in, cannot unmute user"

        #If the user object has a username attribute, use that
        # because most user objects will __str__ into their base 36 ID
        if hasattr(user, "username"):
            user = user.username

        record_id = self.scraper.get_muted_user_record(str(user))
        assert record_id, "User was not in muted records"
        return self.servicephp.unmute_user(record_id)

    def __next_event_json(self):
        """Wait for the next event from the SSE and parse the JSON"""
        if not self.chat_running: #Do not try to query a new event if chat is closed
            print("Chat closed, cannot retrieve new JSON data.")
            return

        try:
            event = next(self.event_generator, None)
        except requests.exceptions.ReadTimeout:
            event = None

        if not event:
            self.chat_running = False #Chat has been closed
            print("Chat has closed.")
            return
        if not event.data: #Blank SSE event
            print("Blank SSE event:>", event, "<:")
            #Self recursion should work so long as we don't get dozens of blank events in a row
            return self.__next_event_json()

        return json.loads(event.data)

    def parse_init_data(self, jsondata):
        """Extract initial chat data from the SSE init event JSON

    Args:
        jsondata (dict): The JSON data returned by the initial SSE connection.
        """

        if jsondata["type"] != "init":
            print(jsondata)
            raise ValueError("That is not init json")

        #Parse pre-connection users, channels, then messages
        self.update_users(jsondata)
        self.update_channels(jsondata)
        self.update_mailbox(jsondata)

        #Load the chat badges
        self.load_badges(jsondata)

        self.rants_enabled = jsondata["data"]["config"]["rants"]["enable"]
        #subscription TODO
        #rant levels TODO
        self.message_length_max = jsondata["data"]["config"]["message_length_max"]

    def update_mailbox(self, jsondata):
        """Parse chat messages from an SSE data JSON

    Args:
        jsondata (dict): A JSON data block from an SSE event.
        """

        #Add new messages
        self.__mailbox += [ChatAPIMessage(message_json, self) for message_json in jsondata["data"]["messages"] if int(message_json["id"]) not in self.__mailbox]

    def clear_mailbox(self):
        """Delete anything in the mailbox"""
        self.__mailbox = []

    def update_users(self, jsondata):
        """Update our dictionary of users from an SSE data JSON

    Args:
        jsondata (dict): A JSON data block from an SSE event.
        """

        for user_json in jsondata["data"]["users"]:
            try:
                self.users[int(user_json["id"])]._jsondata = user_json #Update an existing user's JSON
            except KeyError: #User is new
                self.users[int(user_json["id"])] = ChatAPIUser(user_json, self)

    def update_channels(self, jsondata):
        """Update our dictionary of channels from an SSE data JSON

    Args:
        jsondata (dict): A JSON data block from an SSE event.
        """

        for channel_json in jsondata["data"]["channels"]:
            try:
                self.channels[int(channel_json["id"])]._jsondata = channel_json #Update an existing channel's JSON
            except KeyError: #Channel is new
                self.channels.update({int(channel_json["id"]) : ChatAPIChannel(channel_json, self)})

    def load_badges(self, jsondata):
        """Create our dictionary of badges from an SSE data JSON

    Args:
        jsondata (dict): A JSON data block from an SSE event.
        """

        self.badges = {badge_slug : ChatAPIUserBadge(badge_slug, jsondata["data"]["config"]["badges"][badge_slug], self) for badge_slug in jsondata["data"]["config"]["badges"].keys()}

    @property
    def stream_id_b10(self):
        """The chat ID in use"""
        return utils.base_36_to_10(self.stream_id)

    def get_message(self):
        """Return the next chat message (parsing any additional data), waits for it to come in, returns None if chat closed"""
        #We don't already have messages
        while not self.__mailbox:
            jsondata = self.__next_event_json()

            #The chat has closed
            if not jsondata:
                return

            #Messages were deleted
            if jsondata["type"] in ("delete_messages", "delete_non_rant_messages"):
                #Flag the messages in our history as being deleted
                for message in self.__history:
                    if message.message_id in jsondata["data"]["message_ids"]:
                        message.deleted = True


            #Re-initialize (could contain new messages)
            elif jsondata["type"] == "init":
                self.parse_init_data(jsondata)

            #Pinned message
            elif jsondata["type"] == "pin_message":
                self.pinned_message = ChatAPIMessage(jsondata["data"]["message"], self)

            #New messages
            elif jsondata["type"] == "messages":
                #Parse users, channels, then messages
                self.update_users(jsondata)
                self.update_channels(jsondata)
                self.update_mailbox(jsondata)

            #Unimplemented event type
            else:
                print("API sent an unimplemented SSE event type")
                print(jsondata)

        m = self.__mailbox.pop(0) #Get the oldest message in the mailbox
        self.__history.append(m) #Add the message to the history

        #Make sure the history is not too long, clipping off the oldest messages
        del self.__history[ : max((len(self.__history) - self.history_len, 0))]

        #Return the next message from the mailbox
        return m

history property

The chat history, trimmed to history_len

session_cookie property

The session cookie we are logged in with

stream_id_b10 property

The chat ID in use

__init__(stream_id, username=None, password=None, session=None, history_len=1000)

The Rumble internal chat API

Parameters:

Name	Type	Description	Default
stream_id	(int, str)	Stream ID in base 10 or 36.	required
username	str	Username to login with. Defaults to no login.	None
password	str	Password to log in with. Defaults to no login.	None
session	(str, dict)	Session token or cookie dict to authenticate with. Defaults to getting new session with username and password.	None
history_len	int	Length of message history to store. Defaults to 1000.	1000

Source code in cocorum/chatapi.py

def __init__(self, stream_id, username: str = None, password: str = None, session = None, history_len = 1000):
    """The Rumble internal chat API

Args:
    stream_id (int, str): Stream ID in base 10 or 36.
    username (str): Username to login with.
        Defaults to no login.
    password (str): Password to log in with.
        Defaults to no login.
    session (str, dict): Session token or cookie dict to authenticate with.
        Defaults to getting new session with username and password.
    history_len (int): Length of message history to store.
        Defaults to 1000.
        """

    self.stream_id = utils.ensure_b36(stream_id)

    self.__mailbox = [] #A mailbox if you will
    self.__history = [] #Chat history
    self.history_len = history_len #How many messages to store in history
    self.pinned_message = None #If a message is pinned, it is assigned to this
    self.users = {} #Dictionary of users by user ID
    self.channels = {} #Dictionary of channels by channel ID
    self.badges = {}

    #Generate our URLs
    self.sse_url = static.URI.ChatAPI.sse_stream.format(stream_id_b10 = self.stream_id_b10)
    self.message_api_url = static.URI.ChatAPI.message.format(stream_id_b10 = self.stream_id_b10)

    #Connect to SSE stream
    #Note: We do NOT want this request to have a timeout
    response = requests.get(self.sse_url, stream = True, headers = static.RequestHeaders.sse_api)
    self.client = sseclient.SSEClient(response)
    self.event_generator = self.client.events()
    self.chat_running = True
    self.parse_init_data(self.__next_event_json())

    #If we have session login, use them
    if (username and password) or session:
        self.servicephp = ServicePHP(username, password, session)
        self.scraper = scraping.Scraper(self.servicephp)
    else:
        self.servicephp = None

    #The last time we sent a message
    self.last_send_time = 0

__next_event_json()

Wait for the next event from the SSE and parse the JSON

Source code in cocorum/chatapi.py

def __next_event_json(self):
    """Wait for the next event from the SSE and parse the JSON"""
    if not self.chat_running: #Do not try to query a new event if chat is closed
        print("Chat closed, cannot retrieve new JSON data.")
        return

    try:
        event = next(self.event_generator, None)
    except requests.exceptions.ReadTimeout:
        event = None

    if not event:
        self.chat_running = False #Chat has been closed
        print("Chat has closed.")
        return
    if not event.data: #Blank SSE event
        print("Blank SSE event:>", event, "<:")
        #Self recursion should work so long as we don't get dozens of blank events in a row
        return self.__next_event_json()

    return json.loads(event.data)

clear_mailbox()

Delete anything in the mailbox

Source code in cocorum/chatapi.py

def clear_mailbox(self):
    """Delete anything in the mailbox"""
    self.__mailbox = []

command(command_message)

Send a native chat command

Parameters:

Name	Type	Description	Default
command_message	str	The message you would send to launch this command in chat.	required

Returns:

Name	Type	Description
JSON	dict	The JSON returned by the command.

Source code in cocorum/chatapi.py

def command(self, command_message: str):
    """Send a native chat command

Args:
    command_message (str): The message you would send to launch this command in chat.

Returns:
    JSON (dict): The JSON returned by the command.
    """

    assert command_message.startswith(static.Message.command_prefix), "Not a command message"
    r = requests.post(
        static.URI.ChatAPI.command,
        data = {
            "video_id" : self.stream_id_b10,
            "message" : command_message,
            },
        cookies = self.session_cookie,
        headers = static.RequestHeaders.user_agent,
        timeout = static.Delays.request_timeout,
        )
    assert r.status_code == 200, f"Command failed: {r}\n{r.text}"
    return r.json()

delete_message(message)

Delete a message in chat.

Parameters:

Name	Type	Description	Default
message	int	Object which when converted to integer is the target message ID.	required

Source code in cocorum/chatapi.py

def delete_message(self, message):
    """Delete a message in chat.

Args:
    message (int): Object which when converted to integer is the target message ID.
    """

    assert self.session_cookie, "Not logged in, cannot delete message"
    assert utils.options_check(self.message_api_url + f"/{int(message)}", "DELETE"), "Rumble denied options request to delete message"

    r = requests.delete(
        self.message_api_url + f"/{int(message)}",
        cookies = self.session_cookie,
        # headers = static.RequestHeaders.user_agent,
        timeout = static.Delays.request_timeout,
        )

    if r.status_code != 200:
        print("Error: Deleting message failed,", r, r.content.decode(static.Misc.text_encoding))
        return False

    return True

get_message()

Return the next chat message (parsing any additional data), waits for it to come in, returns None if chat closed

Source code in cocorum/chatapi.py

def get_message(self):
    """Return the next chat message (parsing any additional data), waits for it to come in, returns None if chat closed"""
    #We don't already have messages
    while not self.__mailbox:
        jsondata = self.__next_event_json()

        #The chat has closed
        if not jsondata:
            return

        #Messages were deleted
        if jsondata["type"] in ("delete_messages", "delete_non_rant_messages"):
            #Flag the messages in our history as being deleted
            for message in self.__history:
                if message.message_id in jsondata["data"]["message_ids"]:
                    message.deleted = True


        #Re-initialize (could contain new messages)
        elif jsondata["type"] == "init":
            self.parse_init_data(jsondata)

        #Pinned message
        elif jsondata["type"] == "pin_message":
            self.pinned_message = ChatAPIMessage(jsondata["data"]["message"], self)

        #New messages
        elif jsondata["type"] == "messages":
            #Parse users, channels, then messages
            self.update_users(jsondata)
            self.update_channels(jsondata)
            self.update_mailbox(jsondata)

        #Unimplemented event type
        else:
            print("API sent an unimplemented SSE event type")
            print(jsondata)

    m = self.__mailbox.pop(0) #Get the oldest message in the mailbox
    self.__history.append(m) #Add the message to the history

    #Make sure the history is not too long, clipping off the oldest messages
    del self.__history[ : max((len(self.__history) - self.history_len, 0))]

    #Return the next message from the mailbox
    return m

load_badges(jsondata)

Create our dictionary of badges from an SSE data JSON

Parameters:

Name	Type	Description	Default
jsondata	dict	A JSON data block from an SSE event.	required

Source code in cocorum/chatapi.py

def load_badges(self, jsondata):
    """Create our dictionary of badges from an SSE data JSON

Args:
    jsondata (dict): A JSON data block from an SSE event.
    """

    self.badges = {badge_slug : ChatAPIUserBadge(badge_slug, jsondata["data"]["config"]["badges"][badge_slug], self) for badge_slug in jsondata["data"]["config"]["badges"].keys()}

mute_user(user, duration=None, total=False)

Mute a user.

Parameters:

Name	Type	Description	Default
user	str	Username to mute.	required
duration	int	How long to mute the user in seconds. Defaults to infinite.	None
total	bool	Wether or not they are muted across all videos. Defaults to False, just this video.	False

Source code in cocorum/chatapi.py

def mute_user(self, user, duration: int = None, total: bool = False):
    """Mute a user.

Args:
    user (str): Username to mute.
    duration (int): How long to mute the user in seconds.
        Defaults to infinite.
    total (bool): Wether or not they are muted across all videos.
        Defaults to False, just this video.
        """

    assert self.session_cookie, "Not logged in, cannot mute user"
    return self.servicephp.mute_user(
        username = str(user),
        is_channel = False,
        video = self.stream_id_b10,
        duration = duration,
        total = total
        )

parse_init_data(jsondata)

Extract initial chat data from the SSE init event JSON

Parameters:

Name	Type	Description	Default
jsondata	dict	The JSON data returned by the initial SSE connection.	required

Source code in cocorum/chatapi.py

def parse_init_data(self, jsondata):
    """Extract initial chat data from the SSE init event JSON

Args:
    jsondata (dict): The JSON data returned by the initial SSE connection.
    """

    if jsondata["type"] != "init":
        print(jsondata)
        raise ValueError("That is not init json")

    #Parse pre-connection users, channels, then messages
    self.update_users(jsondata)
    self.update_channels(jsondata)
    self.update_mailbox(jsondata)

    #Load the chat badges
    self.load_badges(jsondata)

    self.rants_enabled = jsondata["data"]["config"]["rants"]["enable"]
    #subscription TODO
    #rant levels TODO
    self.message_length_max = jsondata["data"]["config"]["message_length_max"]

pin_message(message)

Pin a message

Source code in cocorum/chatapi.py

def pin_message(self, message):
    """Pin a message"""
    assert self.session_cookie, "Not logged in, cannot pin message"
    return self.servicephp.chat_pin(self.stream_id_b10, message)

send_message(text, channel_id=None)

Send a message in chat.

Parameters:

Name	Type	Description	Default
text	str	The message text.	required
channel_id	int	Numeric ID of the channel to use. Defaults to None.	None

Returns:

Name	Type	Description
ID	int	The ID of the sent message.
User	ChatAPIUser	Your current chat user information.

Source code in cocorum/chatapi.py

def send_message(self, text: str, channel_id: int = None):
    """Send a message in chat.

Args:
    text (str): The message text.
    channel_id (int): Numeric ID of the channel to use.
        Defaults to None.

Returns:
    ID (int): The ID of the sent message.
    User (ChatAPIUser): Your current chat user information.
    """

    assert self.session_cookie, "Not logged in, cannot send message"
    assert len(text) <= static.Message.max_len, "Mesage is too long"
    curtime = time.time()
    assert self.last_send_time + static.Message.send_cooldown <= curtime, "Sending messages too fast"
    assert utils.options_check(self.message_api_url, "POST"), "Rumble denied options request to post message"
    r = requests.post(
        self.message_api_url,
        cookies = self.session_cookie,
        json = {
            "data": {
                "request_id": utils.generate_request_id(),
                "message": {
                    "text": text
                },
                "rant": None,
                "channel_id": channel_id
                }
            },
        # headers = static.RequestHeaders.user_agent,
        timeout = static.Delays.request_timeout,
        )

    if r.status_code != 200:
        print("Error: Sending message failed,", r, r.text)
        return

    return int(r.json()["data"]["id"]), ChatAPIUser(r.json()["data"]["user"], self)

unmute_user(user)

Unmute a user.

Parameters:

Name	Type	Description	Default
user	str	Username to unmute	required

Source code in cocorum/chatapi.py

def unmute_user(self, user):
    """Unmute a user.

Args:
    user (str): Username to unmute
    """

    assert self.session_cookie, "Not logged in, cannot unmute user"

    #If the user object has a username attribute, use that
    # because most user objects will __str__ into their base 36 ID
    if hasattr(user, "username"):
        user = user.username

    record_id = self.scraper.get_muted_user_record(str(user))
    assert record_id, "User was not in muted records"
    return self.servicephp.unmute_user(record_id)

unpin_message(message=None)

Unpin the pinned message

Source code in cocorum/chatapi.py

def unpin_message(self, message = None):
    """Unpin the pinned message"""
    assert self.session_cookie, "Not logged in, cannot unpin message"
    if not message:
        message = self.pinned_message
    assert message, "No known pinned message and ID not provided"
    return self.servicephp.chat_pin(self.stream_id_b10, message, unpin = True)

update_channels(jsondata)

Update our dictionary of channels from an SSE data JSON

Parameters:

Name	Type	Description	Default
jsondata	dict	A JSON data block from an SSE event.	required

Source code in cocorum/chatapi.py

def update_channels(self, jsondata):
    """Update our dictionary of channels from an SSE data JSON

Args:
    jsondata (dict): A JSON data block from an SSE event.
    """

    for channel_json in jsondata["data"]["channels"]:
        try:
            self.channels[int(channel_json["id"])]._jsondata = channel_json #Update an existing channel's JSON
        except KeyError: #Channel is new
            self.channels.update({int(channel_json["id"]) : ChatAPIChannel(channel_json, self)})

update_mailbox(jsondata)

Parse chat messages from an SSE data JSON

Parameters:

Name	Type	Description	Default
jsondata	dict	A JSON data block from an SSE event.	required

Source code in cocorum/chatapi.py

def update_mailbox(self, jsondata):
    """Parse chat messages from an SSE data JSON

Args:
    jsondata (dict): A JSON data block from an SSE event.
    """

    #Add new messages
    self.__mailbox += [ChatAPIMessage(message_json, self) for message_json in jsondata["data"]["messages"] if int(message_json["id"]) not in self.__mailbox]

update_users(jsondata)

Update our dictionary of users from an SSE data JSON

Parameters:

Name	Type	Description	Default
jsondata	dict	A JSON data block from an SSE event.	required

Source code in cocorum/chatapi.py

def update_users(self, jsondata):
    """Update our dictionary of users from an SSE data JSON

Args:
    jsondata (dict): A JSON data block from an SSE event.
    """

    for user_json in jsondata["data"]["users"]:
        try:
            self.users[int(user_json["id"])]._jsondata = user_json #Update an existing user's JSON
        except KeyError: #User is new
            self.users[int(user_json["id"])] = ChatAPIUser(user_json, self)

ChatAPIChannel

Bases: ChatAPIChatter

A channel in the SSE chat

Source code in cocorum/chatapi.py

class ChatAPIChannel(ChatAPIChatter):
    """A channel in the SSE chat"""
    def __init__(self, jsondata, chat):
        """A channel in the internal chat API

    Args:
        jsondata (dict): The JSON data block for the channel.
        chat (ChatAPI): The ChatAPI object that spawned us.
        """

        super().__init__(jsondata, chat)

        #Find the user who has this channel
        for user in self.chat.users.values():
            if user.channel_id == self.channel_id or self.channel_id in user.previous_channel_ids:
                self.user = user
                break

    @property
    def is_appearing(self):
        """Is the user of this channel still appearing as it?"""
        return self.user.channel_id == self.channel_id #The user channel_id still matches our own

    @property
    def channel_id(self):
        """The ID of this channel in base 10"""
        return int(self["id"])

    @property
    def channel_id_b10(self):
        """The ID of this channel in base 10"""
        return self.channel_id

    @property
    def channel_id_b36(self):
        """The ID of this channel in base 36"""
        return utils.base_10_to_36(self.channel_id)

    @property
    def user_id(self):
        """The numeric ID of the user of this channel"""
        return self.user.user_id

    @property
    def user_id_b36(self):
        """The numeric ID of the user of this channel in base 36"""
        return self.user.user_id_b36

    @property
    def user_id_b10(self):
        """The numeric ID of the user of this channel in base 10"""
        return self.user.user_id_b10

channel_id property

The ID of this channel in base 10

channel_id_b10 property

The ID of this channel in base 10

channel_id_b36 property

The ID of this channel in base 36

is_appearing property

Is the user of this channel still appearing as it?

user_id property

The numeric ID of the user of this channel

user_id_b10 property

The numeric ID of the user of this channel in base 10

user_id_b36 property

The numeric ID of the user of this channel in base 36

__init__(jsondata, chat)

A channel in the internal chat API

Parameters:

Name	Type	Description	Default
jsondata	dict	The JSON data block for the channel.	required
chat	ChatAPI	The ChatAPI object that spawned us.	required

Source code in cocorum/chatapi.py

def __init__(self, jsondata, chat):
    """A channel in the internal chat API

Args:
    jsondata (dict): The JSON data block for the channel.
    chat (ChatAPI): The ChatAPI object that spawned us.
    """

    super().__init__(jsondata, chat)

    #Find the user who has this channel
    for user in self.chat.users.values():
        if user.channel_id == self.channel_id or self.channel_id in user.previous_channel_ids:
            self.user = user
            break

ChatAPIChatter

Bases: JSONUserAction, ChatAPIObj

A user or channel in the internal chat API (abstract)

Source code in cocorum/chatapi.py

class ChatAPIChatter(JSONUserAction, ChatAPIObj):
    """A user or channel in the internal chat API (abstract)"""
    def __init__(self, jsondata, chat):
        """A user or channel in the internal chat API (abstract)

    Args:
        jsondata (dict): The JSON data block for the user/channel.
        chat (ChatAPI): The ChatAPI object that spawned us.
        """
        ChatAPIObj.__init__(self, jsondata, chat)
        JSONUserAction.__init__(self, jsondata)

    @property
    def link(self):
        """The user's subpage of Rumble.com"""
        return self["link"]

link property

The user's subpage of Rumble.com

__init__(jsondata, chat)

A user or channel in the internal chat API (abstract)

Parameters:

Name	Type	Description	Default
jsondata	dict	The JSON data block for the user/channel.	required
chat	ChatAPI	The ChatAPI object that spawned us.	required

Source code in cocorum/chatapi.py

def __init__(self, jsondata, chat):
    """A user or channel in the internal chat API (abstract)

Args:
    jsondata (dict): The JSON data block for the user/channel.
    chat (ChatAPI): The ChatAPI object that spawned us.
    """
    ChatAPIObj.__init__(self, jsondata, chat)
    JSONUserAction.__init__(self, jsondata)

ChatAPIMessage

Bases: ChatAPIObj

A single chat message in the internal chat API

Source code in cocorum/chatapi.py

class ChatAPIMessage(ChatAPIObj):
    """A single chat message in the internal chat API"""
    def __init__(self, jsondata, chat):
        """A single chat message in the internal chat API

    Args:
        jsondata (dict): The JSON data block for the message.
        chat (ChatAPI): The ChatAPI object that spawned us.
        """

        super().__init__(jsondata, chat)

        #Set the channel ID of our user if we can
        if self.user:
            self.user._set_channel_id = self.channel_id

        #Remember if we were deleted
        self.deleted = False

    def __eq__(self, other):
        """Compare this chat message with another

    Args:
        other (str, ChatAPIMessage): Object to compare to.

    Returns:
        Comparison (bool, None): Did it fit the criteria?
        """

        if isinstance(other, str):
            return self.text == other

        #Check if the other object's text matches our own, if it has such
        if hasattr(other, "text"):
            #Check if the other object's user ID matches our own, if it has one
            if hasattr(other, "user_id"):
                #Check if the other object is a raid notification, if it says
                if hasattr(other, "raid_notification"):
                    return (self.user_id, self.text, self.raid_notification) == (other.user_id, other.text, other.raid_notification)

                return (self.user_id, self.text) == (other.user_id, other.text)

            #Check if the other object's username matches our own, if it has one
            if hasattr(other, "username"):
                #Check if the other object is a raid notification, if it says
                if hasattr(other, "raid_notification"):
                    return (self.user_id, self.text, self.raid_notification) == (other.user_id, other.text, other.raid_notification)

                return (self.user.username, self.text) == (other.username, other.text)

            #No user identifying attributes, but the text does match
            return self.text == other.text

    def __str__(self):
        """The chat message in string form"""
        return self.text

    def __int__(self):
        """The chat message in integer (ID) form"""
        return self.message_id

    @property
    def message_id(self):
        """The unique numerical ID of the chat message in base 10"""
        return int(self["id"])

    @property
    def message_id_b10(self):
        """The unique numerical ID of the chat message in base 10"""
        return self.message_id

    @property
    def message_id_b36(self):
        """The unique numerical ID of the chat message in base 36"""
        return utils.base_10_to_36(self.message_id)

    @property
    def time(self):
        """The time the message was sent on, in seconds since the Epoch UTC"""
        return utils.parse_timestamp(self["time"])

    @property
    def user_id(self):
        """The numerical ID of the user who posted the message in base 10"""
        return int(self["user_id"])

    @property
    def user_id_b10(self):
        """The numeric ID of the user in base 10"""
        return self.user_id

    @property
    def user_id_b36(self):
        """The numeric ID of the user in base 36"""
        return utils.base_10_to_36(self.user_id)

    @property
    def channel_id(self):
        """The numeric ID of the channel who posted the message, if there is one"""
        try:
            #Note: For some reason, channel IDs in messages alone show up as integers in the SSE events
            return int(self["channel_id"])
        except KeyError: #This user is not appearing as a channel and so has no channel ID
            return None

    @property
    def channel_id_b10(self):
        """The ID of the channel who posted the message in base 10"""
        return self.channel_id

    @property
    def channel_id_b36(self):
        """The ID of the channel who posted the message in base 36"""
        if not self.channel_id:
            return
        return utils.base_10_to_36(self.channel_id)

    @property
    def text(self):
        """The text of the message"""
        return self["text"]

    @property
    def user(self):
        """Reference to the user who posted this message"""
        try:
            return self.chat.users[self.user_id]
        except KeyError:
            print(f"ERROR: Message {self.message_id} could not reference user {self.user_id} because chat has no records of them as of yet.")

    @property
    def channel(self):
        """Reference to the channel that posted this message, if there was one"""
        if not self.channel_id:
            return None

        return self.chat.channels[self.channel_id]

    @property
    def is_rant(self):
        """Is this message a rant?"""
        return "rant" in self._jsondata

    @property
    def rant_price_cents(self):
        """The price of the rant, returns 0 if message is not a rant"""
        if not self.is_rant:
            return 0
        return self["rant"]["price_cents"]

    @property
    def rant_duration(self):
        """The duration the rant will show for, returns 0 if message is not a rant"""
        if not self.is_rant:
            return 0
        return self["rant"]["duration"]

    @property
    def rant_expires_on(self):
        """When the rant expires, returns message creation time if message is not a rant"""
        if not self.is_rant:
            return self.time
        return utils.parse_timestamp(self["rant"]["expires_on"])

    @property
    def raid_notification(self):
        """Are we a raid notification? Returns associated JSON data if yes, False if no"""
        if "raid_notification" in self._jsondata:
            return self["raid_notification"]

        return False

channel property

Reference to the channel that posted this message, if there was one

channel_id property

The numeric ID of the channel who posted the message, if there is one

channel_id_b10 property

The ID of the channel who posted the message in base 10

channel_id_b36 property

The ID of the channel who posted the message in base 36

is_rant property

Is this message a rant?

message_id property

The unique numerical ID of the chat message in base 10

message_id_b10 property

The unique numerical ID of the chat message in base 10

message_id_b36 property

The unique numerical ID of the chat message in base 36

raid_notification property

Are we a raid notification? Returns associated JSON data if yes, False if no

rant_duration property

The duration the rant will show for, returns 0 if message is not a rant

rant_expires_on property

When the rant expires, returns message creation time if message is not a rant

rant_price_cents property

The price of the rant, returns 0 if message is not a rant

text property

The text of the message

time property

The time the message was sent on, in seconds since the Epoch UTC

user property

Reference to the user who posted this message

user_id property

The numerical ID of the user who posted the message in base 10

user_id_b10 property

The numeric ID of the user in base 10

user_id_b36 property

The numeric ID of the user in base 36

__eq__(other)

Compare this chat message with another

Parameters:

Name	Type	Description	Default
other	(str, ChatAPIMessage)	Object to compare to.	required

Returns:

Name	Type	Description
Comparison	(bool, None)	Did it fit the criteria?

Source code in cocorum/chatapi.py

def __eq__(self, other):
    """Compare this chat message with another

Args:
    other (str, ChatAPIMessage): Object to compare to.

Returns:
    Comparison (bool, None): Did it fit the criteria?
    """

    if isinstance(other, str):
        return self.text == other

    #Check if the other object's text matches our own, if it has such
    if hasattr(other, "text"):
        #Check if the other object's user ID matches our own, if it has one
        if hasattr(other, "user_id"):
            #Check if the other object is a raid notification, if it says
            if hasattr(other, "raid_notification"):
                return (self.user_id, self.text, self.raid_notification) == (other.user_id, other.text, other.raid_notification)

            return (self.user_id, self.text) == (other.user_id, other.text)

        #Check if the other object's username matches our own, if it has one
        if hasattr(other, "username"):
            #Check if the other object is a raid notification, if it says
            if hasattr(other, "raid_notification"):
                return (self.user_id, self.text, self.raid_notification) == (other.user_id, other.text, other.raid_notification)

            return (self.user.username, self.text) == (other.username, other.text)

        #No user identifying attributes, but the text does match
        return self.text == other.text

__init__(jsondata, chat)

A single chat message in the internal chat API

Parameters:

Name	Type	Description	Default
jsondata	dict	The JSON data block for the message.	required
chat	ChatAPI	The ChatAPI object that spawned us.	required

Source code in cocorum/chatapi.py

def __init__(self, jsondata, chat):
    """A single chat message in the internal chat API

Args:
    jsondata (dict): The JSON data block for the message.
    chat (ChatAPI): The ChatAPI object that spawned us.
    """

    super().__init__(jsondata, chat)

    #Set the channel ID of our user if we can
    if self.user:
        self.user._set_channel_id = self.channel_id

    #Remember if we were deleted
    self.deleted = False

__int__()

The chat message in integer (ID) form

Source code in cocorum/chatapi.py

def __int__(self):
    """The chat message in integer (ID) form"""
    return self.message_id

__str__()

The chat message in string form

Source code in cocorum/chatapi.py

def __str__(self):
    """The chat message in string form"""
    return self.text

ChatAPIObj

Bases: JSONObj

Object in the internal chat API

Source code in cocorum/chatapi.py

class ChatAPIObj(JSONObj):
    """Object in the internal chat API"""
    def __init__(self, jsondata, chat):
        """Object in the internal chat API

    Args:
        jsondata (dict): The JSON data block for the object.
        chat (ChatAPI): The ChatAPI object that spawned us.
        """

        JSONObj.__init__(self, jsondata)
        self.chat = chat

__init__(jsondata, chat)

Object in the internal chat API

Parameters:

Name	Type	Description	Default
jsondata	dict	The JSON data block for the object.	required
chat	ChatAPI	The ChatAPI object that spawned us.	required

Source code in cocorum/chatapi.py

def __init__(self, jsondata, chat):
    """Object in the internal chat API

Args:
    jsondata (dict): The JSON data block for the object.
    chat (ChatAPI): The ChatAPI object that spawned us.
    """

    JSONObj.__init__(self, jsondata)
    self.chat = chat

ChatAPIUser

Bases: ChatAPIChatter

User in the internal chat API

Source code in cocorum/chatapi.py

class ChatAPIUser(ChatAPIChatter):
    """User in the internal chat API"""
    def __init__(self, jsondata, chat):
        """A user in the internal chat API

    Args:
        jsondata (dict): The JSON data block for the user.
        chat (ChatAPI): The ChatAPI object that spawned us.
        """

        ChatAPIChatter.__init__(self, jsondata, chat)
        self.previous_channel_ids = [] #List of channels the user has appeared as, including the current one
        self._set_channel_id = None #Channel ID set from message

    def __int__(self):
        """The user as an integer (it's ID in base 10)"""
        return self.user_id_b10

    @property
    def user_id(self):
        """The numeric ID of the user in base 10"""
        return int(self["id"])

    @property
    def user_id_b10(self):
        """The numeric ID of the user in base 10"""
        return self.user_id

    @property
    def user_id_b36(self):
        """The numeric ID of the user in base 36"""
        return utils.base_10_to_36(self.user_id)

    @property
    def channel_id(self):
        """The numeric channel ID that the user is appearing with in base 10"""

        #Try to get our channel ID from our own JSON (may be deprecated)
        try:
            new = int(self["channel_id"])

        #Rely on messages to have assigned our channel ID
        except KeyError:
            new = self._set_channel_id

        if new not in self.previous_channel_ids: #Record the appearance of a new chanel appearance, including None
            self.previous_channel_ids.append(new)
        return new

    @property
    def channel_id_b10(self):
        """The numeric channel ID that the user is appearing with in base 10"""
        return self.channel_id

    @property
    def channel_id_b36(self):
        """The numeric channel ID that the user is appearing with in base 36"""
        if not self.channel_id:
            return
        return utils.base_10_to_36(self.channel_id)

    @property
    def is_follower(self):
        """Is this user following the livestreaming channel?"""
        return self["is_follower"]

    @property
    def color(self):
        """The color of our username (RGB tuple)"""
        return tuple(int(self["color"][i : i + 2], 16) for i in range(0, 6, 2))

    @property
    def badges(self):
        """Badges the user has"""
        try:
            return [self.chat.badges[badge_slug] for badge_slug in self["badges"]]

        #User has no badges
        except KeyError:
            return []

badges property

Badges the user has

channel_id property

The numeric channel ID that the user is appearing with in base 10

channel_id_b10 property

The numeric channel ID that the user is appearing with in base 10

channel_id_b36 property

The numeric channel ID that the user is appearing with in base 36

color property

The color of our username (RGB tuple)

is_follower property

Is this user following the livestreaming channel?

user_id property

The numeric ID of the user in base 10

user_id_b10 property

The numeric ID of the user in base 10

user_id_b36 property

The numeric ID of the user in base 36

__init__(jsondata, chat)

A user in the internal chat API

Parameters:

Name	Type	Description	Default
jsondata	dict	The JSON data block for the user.	required
chat	ChatAPI	The ChatAPI object that spawned us.	required

Source code in cocorum/chatapi.py

def __init__(self, jsondata, chat):
    """A user in the internal chat API

Args:
    jsondata (dict): The JSON data block for the user.
    chat (ChatAPI): The ChatAPI object that spawned us.
    """

    ChatAPIChatter.__init__(self, jsondata, chat)
    self.previous_channel_ids = [] #List of channels the user has appeared as, including the current one
    self._set_channel_id = None #Channel ID set from message

__int__()

The user as an integer (it's ID in base 10)

Source code in cocorum/chatapi.py

def __int__(self):
    """The user as an integer (it's ID in base 10)"""
    return self.user_id_b10

ChatAPIUserBadge

Bases: ChatAPIObj

A badge of a user

Source code in cocorum/chatapi.py

class ChatAPIUserBadge(ChatAPIObj):
    """A badge of a user"""
    def __init__(self, slug, jsondata, chat):
        """A user badge in the internal chat API

    Args:
        jsondata (dict): The JSON data block for the user badge.
        chat (ChatAPI): The ChatAPI object that spawned us.
        """

        super().__init__(jsondata, chat)
        self.slug = slug #The unique identification for this badge
        self.__icon = None

    def __eq__(self, other):
        """Check if this badge is equal to another

    Args:
        other (str, ChatAPIUserBadge): Object to compare to.

    Returns:
        Comparison (bool, None): Did it fit the criteria?
        """

        #Check if the string is either our slug or our label in any language
        if isinstance(other, str):
            return other in (self.slug, self.label.values())

        #Check if the compared object has the same slug, if it has one
        if hasattr(other, "slug"):
            return self.slug == other.slug

    def __str__(self):
        """The chat user badge in string form"""
        return self.slug

    @property
    def label(self):
        """A dictionary of lang:label pairs"""
        return self["label"]

    @property
    def icon_url(self):
        """The URL of the badge's icon"""
        return static.URI.rumble_base + self["icons"][static.Misc.badge_icon_size]

    @property
    def icon(self):
        """The badge's icon as a bytestring"""
        if not self.__icon: #We never queried the icon before
            response = requests.get(self.icon_url, timeout = static.Delays.request_timeout)
            assert response.status_code == 200, "Status code " + str(response.status_code)

            self.__icon = response.content

        return self.__icon

icon property

The badge's icon as a bytestring

icon_url property

The URL of the badge's icon

label property

A dictionary of lang:label pairs

__eq__(other)

Check if this badge is equal to another

Parameters:

Name	Type	Description	Default
other	(str, ChatAPIUserBadge)	Object to compare to.	required

Returns:

Name	Type	Description
Comparison	(bool, None)	Did it fit the criteria?

Source code in cocorum/chatapi.py

def __eq__(self, other):
    """Check if this badge is equal to another

Args:
    other (str, ChatAPIUserBadge): Object to compare to.

Returns:
    Comparison (bool, None): Did it fit the criteria?
    """

    #Check if the string is either our slug or our label in any language
    if isinstance(other, str):
        return other in (self.slug, self.label.values())

    #Check if the compared object has the same slug, if it has one
    if hasattr(other, "slug"):
        return self.slug == other.slug

__init__(slug, jsondata, chat)

A user badge in the internal chat API

Parameters:

Name	Type	Description	Default
jsondata	dict	The JSON data block for the user badge.	required
chat	ChatAPI	The ChatAPI object that spawned us.	required

Source code in cocorum/chatapi.py

def __init__(self, slug, jsondata, chat):
    """A user badge in the internal chat API

Args:
    jsondata (dict): The JSON data block for the user badge.
    chat (ChatAPI): The ChatAPI object that spawned us.
    """

    super().__init__(jsondata, chat)
    self.slug = slug #The unique identification for this badge
    self.__icon = None

__str__()

The chat user badge in string form

Source code in cocorum/chatapi.py

def __str__(self):
    """The chat user badge in string form"""
    return self.slug

S.D.G.