cocorum.chatapi

The primary use from this module is the ChatAPI class, a wrapper for Rumble's internal chat API. This class includes links to the chat-related methods of ServicePHP() (muting users for example). If you wish to use any two-way chat features, you must first create an instance of cocorum.servicephp.ServicePHP(), and then pass it to this class upon initialization. All other classes are supporting sub-classes.

Internal chat API client

Interface with the Rumble chat API to send and receive messages, etc.

Copyright 2026 Wilbur Jaywright d.b.a. Marswide BGL.

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

S.D.G.

BaseComment

A comment on a Rumble video

Source code in cocorum/basehandles.py

class BaseComment:
    """A comment on a Rumble video"""

    def __int__(self) -> int:
        """The comment in integer form (its ID)"""
        return self.comment_id

    def __str__(self) -> str:
        """The comment as a string (its text)"""
        return self.text

    @property
    def comment_id_b10(self) -> int:
        """The base 10 ID of the comment"""
        assert isinstance(
            self.comment_id, int), "Contact the Cocorum developers, this is probably their mistake"
        return self.comment_id

    @property
    def comment_id_b36(self) -> str:
        """The base 36 ID of the comment"""
        return utils.base_10_to_36(self.comment_id)

    def __eq__(self, other: Any) -> bool:
        """Determine if this comment is equal to another.

        Args:
            other (Any): Object to compare to.

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

        # Check for direct matches first
        if isinstance(other, int):
            return self.comment_id_b10 == other
        if isinstance(other, str):
            return str(self) == other

        # Check for object attributes to match to
        if hasattr(other, "comment_id_b10"):
            return self.comment_id_b10 == other.comment_id_b10

        # Check conversion to integer last
        if hasattr(other, "__int__"):
            return self.comment_id_b10 == int(other)

        return False

    def pin(self, unpin: bool = False):
        """Pin or unpin this comment.

    Args:
        unpin (bool): If true, unpins instead of pinning comment.
        """

        return self.servicephp.comment_pin(self, unpin)

    def delete(self):
        """Delete this comment"""

        return self.servicephp.comment_delete(self)

    def restore(self) -> APIComment:
        """Un-delete this comment"""

        return self.servicephp.comment_restore(self)

comment_id_b10 property

The base 10 ID of the comment

comment_id_b36 property

The base 36 ID of the comment

__eq__(other)

Determine if this comment is equal to another.

Parameters:

Name	Type	Description	Default
other	Any	Object to compare to.	required

Returns:

Name	Type	Description
Comparison	bool	Did it fit the criteria?

Source code in cocorum/basehandles.py

def __eq__(self, other: Any) -> bool:
    """Determine if this comment is equal to another.

    Args:
        other (Any): Object to compare to.

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

    # Check for direct matches first
    if isinstance(other, int):
        return self.comment_id_b10 == other
    if isinstance(other, str):
        return str(self) == other

    # Check for object attributes to match to
    if hasattr(other, "comment_id_b10"):
        return self.comment_id_b10 == other.comment_id_b10

    # Check conversion to integer last
    if hasattr(other, "__int__"):
        return self.comment_id_b10 == int(other)

    return False

__int__()

The comment in integer form (its ID)

Source code in cocorum/basehandles.py

def __int__(self) -> int:
    """The comment in integer form (its ID)"""
    return self.comment_id

__str__()

The comment as a string (its text)

Source code in cocorum/basehandles.py

def __str__(self) -> str:
    """The comment as a string (its text)"""
    return self.text

delete()

Delete this comment

Source code in cocorum/basehandles.py

def delete(self):
    """Delete this comment"""

    return self.servicephp.comment_delete(self)

pin(unpin=False)

Pin or unpin this comment.

Parameters:

Name	Type	Description	Default
unpin	bool	If true, unpins instead of pinning comment.	False

Source code in cocorum/basehandles.py

def pin(self, unpin: bool = False):
    """Pin or unpin this comment.

Args:
    unpin (bool): If true, unpins instead of pinning comment.
    """

    return self.servicephp.comment_pin(self, unpin)

restore()

Un-delete this comment

Source code in cocorum/basehandles.py

def restore(self) -> APIComment:
    """Un-delete this comment"""

    return self.servicephp.comment_restore(self)

BaseContentVotes

Likes and dislikes on a video or comment

Source code in cocorum/basehandles.py

class BaseContentVotes:
    """Likes and dislikes on a video or comment"""

    def __repr__(self) -> str:
        """String to represent this object"""
        return f"{type(self).__name__}(content_id={self.content_id}, score={self.score})"

    def __int__(self) -> int:
        """The integer form of the content votes"""
        return self.score

    def __eq__(self, other: Any) -> bool:
        """Determine if this content votes is equal to another.

        Args:
            other (Any): Object to compare to.

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

        # Check for direct matches first
        if isinstance(other, int):
            return self.score == other
        if isinstance(other, str):
            return str(self) == other

        # Check for object attributes to match to
        if hasattr(other, "score"):
            return self.score == other.score

        # Check conversion to integer last
        if hasattr(other, "__int__"):
            return self.score == int(other)

        return False

__eq__(other)

Determine if this content votes is equal to another.

Parameters:

Name	Type	Description	Default
other	Any	Object to compare to.	required

Returns:

Name	Type	Description
Comparison	bool	Did it fit the criteria?

Source code in cocorum/basehandles.py

def __eq__(self, other: Any) -> bool:
    """Determine if this content votes is equal to another.

    Args:
        other (Any): Object to compare to.

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

    # Check for direct matches first
    if isinstance(other, int):
        return self.score == other
    if isinstance(other, str):
        return str(self) == other

    # Check for object attributes to match to
    if hasattr(other, "score"):
        return self.score == other.score

    # Check conversion to integer last
    if hasattr(other, "__int__"):
        return self.score == int(other)

    return False

__int__()

The integer form of the content votes

Source code in cocorum/basehandles.py

def __int__(self) -> int:
    """The integer form of the content votes"""
    return self.score

__repr__()

String to represent this object

Source code in cocorum/basehandles.py

def __repr__(self) -> str:
    """String to represent this object"""
    return f"{type(self).__name__}(content_id={self.content_id}, score={self.score})"

BasePlaylist

A playlist of Rumble videos

Source code in cocorum/basehandles.py

class BasePlaylist:
    """A playlist of Rumble videos"""

    def __int__(self) -> int:
        """The playlist as an integer (it's ID in base 10)"""
        return self.playlist_id_b10

    def __str__(self) -> str:
        """The playlist as a string (it's ID in base 64)"""
        return self.playlist_id_b64

    def __repr__(self) -> str:
        """String to represent this object"""
        return f"{type(self).__name__}(playlist_id={self.playlist_id}, title=\"{self.title}\")"

    def __eq__(self, other: Any) -> bool:
        """Determine if this playlist is equal to another.

        Args:
            other (Any): Object to compare to.

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

        # Check for direct matches first
        if isinstance(other, int):
            return self.playlist_id_b64 == other
        if isinstance(other, str):
            return str(other) == self.playlist_id_b64

        # # Check for object attributes to match to
        # if hasattr(other, "playlist_id_b10"):
        #     return self.playlist_id_b10 == other.playlist_id_b10

        # # Check conversion to integer last, in case another ID or something happens to match
        # if hasattr(other, "__int__"):
        #     return self.playlist_id_b10 == int(other)

        return False

    @property
    def playlist_id_b64(self) -> str:
        """The numeric ID of the playlist in base 64"""
        return self.playlist_id

    @property
    def playlist_id_b10(self) -> int:
        """The numeric ID of the playlist in base 10"""
        raise NotImplementedError("See Cocorum issue #22")
        # return utils.base_64_to_10(self.playlist_id)

    def add_video(self, video_id: SupportsInt | str):
        """Add a video to this playlist

    Args:
        video_id (SupportsInt | str): The numeric ID of the video to add, in base 10 or 36.
        """

        return self.servicephp.playlist_add_video(self.playlist_id, video_id)

    def delete_video(self, video_id: SupportsInt | str):
        """Remove a video from this playlist

    Args:
        video_id (SupportsInt | str): The numeric ID of the video to remove, in base 10 or 36.
        """

        self.servicephp.playlist_delete_video(self.playlist_id, video_id)

    def edit(self, title: str = None, description: str = None, visibility: str = None, channel_id: Optional[SupportsInt | str] = None) -> APIPlaylist:
        """Edit the details of this playlist. WARNING: The original object will probably be stale after this operation.

        Args:
            title (str): The title of the playlist.
                Defaults to staying the same.
            description (str): Describe the playlist.
                Defaults to staying the same.
            visibility (str): Set to public, unlisted, or private via string.
                Defaults to staying the same.
            channel_id (SupportsInt | str): The ID of the channel to have the playlist under. TODO: Cannot be retrieved!
                Defaults to resetting to None.

        Returns:
            playlist (APIPlaylist): The edit result.
        """

        if title is None:
            title = self.title
        if description is None:
            description = self.description
        if visibility is None:
            visibility = self.visibility
        # if channel_id is False:
        #     channel_id = self.channel_id

        return self.servicephp.playlist_edit(self.playlist_id, title, description, visibility, channel_id)

    def delete(self):
        """Delete this playlist"""

        self.servicephp.playlist_delete(self)

playlist_id_b10 property

The numeric ID of the playlist in base 10

playlist_id_b64 property

The numeric ID of the playlist in base 64

__eq__(other)

Determine if this playlist is equal to another.

Parameters:

Name	Type	Description	Default
other	Any	Object to compare to.	required

Returns:

Name	Type	Description
Comparison	bool	Did it fit the criteria?

Source code in cocorum/basehandles.py

def __eq__(self, other: Any) -> bool:
    """Determine if this playlist is equal to another.

    Args:
        other (Any): Object to compare to.

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

    # Check for direct matches first
    if isinstance(other, int):
        return self.playlist_id_b64 == other
    if isinstance(other, str):
        return str(other) == self.playlist_id_b64

    # # Check for object attributes to match to
    # if hasattr(other, "playlist_id_b10"):
    #     return self.playlist_id_b10 == other.playlist_id_b10

    # # Check conversion to integer last, in case another ID or something happens to match
    # if hasattr(other, "__int__"):
    #     return self.playlist_id_b10 == int(other)

    return False

__int__()

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

Source code in cocorum/basehandles.py

def __int__(self) -> int:
    """The playlist as an integer (it's ID in base 10)"""
    return self.playlist_id_b10

__repr__()

String to represent this object

Source code in cocorum/basehandles.py

def __repr__(self) -> str:
    """String to represent this object"""
    return f"{type(self).__name__}(playlist_id={self.playlist_id}, title=\"{self.title}\")"

__str__()

The playlist as a string (it's ID in base 64)

Source code in cocorum/basehandles.py

def __str__(self) -> str:
    """The playlist as a string (it's ID in base 64)"""
    return self.playlist_id_b64

add_video(video_id)

Add a video to this playlist

Parameters:

Name	Type	Description	Default
video_id	SupportsInt | str	The numeric ID of the video to add, in base 10 or 36.	required

Source code in cocorum/basehandles.py

def add_video(self, video_id: SupportsInt | str):
    """Add a video to this playlist

Args:
    video_id (SupportsInt | str): The numeric ID of the video to add, in base 10 or 36.
    """

    return self.servicephp.playlist_add_video(self.playlist_id, video_id)

delete()

Delete this playlist

Source code in cocorum/basehandles.py

def delete(self):
    """Delete this playlist"""

    self.servicephp.playlist_delete(self)

delete_video(video_id)

Remove a video from this playlist

Parameters:

Name	Type	Description	Default
video_id	SupportsInt | str	The numeric ID of the video to remove, in base 10 or 36.	required

Source code in cocorum/basehandles.py

def delete_video(self, video_id: SupportsInt | str):
    """Remove a video from this playlist

Args:
    video_id (SupportsInt | str): The numeric ID of the video to remove, in base 10 or 36.
    """

    self.servicephp.playlist_delete_video(self.playlist_id, video_id)

edit(title=None, description=None, visibility=None, channel_id=None)

Edit the details of this playlist. WARNING: The original object will probably be stale after this operation.

Parameters:

Name	Type	Description	Default
title	str	The title of the playlist. Defaults to staying the same.	None
description	str	Describe the playlist. Defaults to staying the same.	None
visibility	str	Set to public, unlisted, or private via string. Defaults to staying the same.	None
channel_id	SupportsInt | str	The ID of the channel to have the playlist under. TODO: Cannot be retrieved! Defaults to resetting to None.	None

Returns:

Name	Type	Description
playlist	APIPlaylist	The edit result.

Source code in cocorum/basehandles.py

def edit(self, title: str = None, description: str = None, visibility: str = None, channel_id: Optional[SupportsInt | str] = None) -> APIPlaylist:
    """Edit the details of this playlist. WARNING: The original object will probably be stale after this operation.

    Args:
        title (str): The title of the playlist.
            Defaults to staying the same.
        description (str): Describe the playlist.
            Defaults to staying the same.
        visibility (str): Set to public, unlisted, or private via string.
            Defaults to staying the same.
        channel_id (SupportsInt | str): The ID of the channel to have the playlist under. TODO: Cannot be retrieved!
            Defaults to resetting to None.

    Returns:
        playlist (APIPlaylist): The edit result.
    """

    if title is None:
        title = self.title
    if description is None:
        description = self.description
    if visibility is None:
        visibility = self.visibility
    # if channel_id is False:
    #     channel_id = self.channel_id

    return self.servicephp.playlist_edit(self.playlist_id, title, description, visibility, channel_id)

BaseUser

A Rumble user

Source code in cocorum/basehandles.py

class BaseUser:
    """A Rumble user"""

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

    def __eq__(self, other: Any) -> bool:
        """Determine if this user is equal to another.

        Args:
            other (Any): Object to compare to.

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

        # Check for direct matches first
        if isinstance(other, int):
            return self.user_id_b10 == other
        if isinstance(other, str):
            return str(other) in (self.user_id_b36, self.username)

        # Check for object attributes to match to
        if hasattr(other, "user_id"):
            return self.user_id_b10 == utils.ensure_b10(other.user_id)

        # Check conversion to integer last, in case another ID or something happens to match
        if hasattr(other, "__int__"):
            return self.user_id_b10 == int(other)

        return False

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

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

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

    Args:
        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.
            """

        self.servicephp.mute(self, self.username, duration, total)

    def unmute(self):
        """Unmute this user."""
        self.servicephp.unmute(self.username)

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)

Determine if this user is equal to another.

Parameters:

Name	Type	Description	Default
other	Any	Object to compare to.	required

Returns:

Name	Type	Description
Comparison	bool	Did it fit the criteria?

Source code in cocorum/basehandles.py

def __eq__(self, other: Any) -> bool:
    """Determine if this user is equal to another.

    Args:
        other (Any): Object to compare to.

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

    # Check for direct matches first
    if isinstance(other, int):
        return self.user_id_b10 == other
    if isinstance(other, str):
        return str(other) in (self.user_id_b36, self.username)

    # Check for object attributes to match to
    if hasattr(other, "user_id"):
        return self.user_id_b10 == utils.ensure_b10(other.user_id)

    # Check conversion to integer last, in case another ID or something happens to match
    if hasattr(other, "__int__"):
        return self.user_id_b10 == int(other)

    return False

__int__()

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

Source code in cocorum/basehandles.py

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

mute(duration=None, total=False)

Mute this user.

Parameters:

Name	Type	Description	Default
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/basehandles.py

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

Args:
    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.
        """

    self.servicephp.mute(self, self.username, duration, total)

unmute()

Unmute this user.

Source code in cocorum/basehandles.py

def unmute(self):
    """Unmute this user."""
    self.servicephp.unmute(self.username)

BaseUserBadge

A badge on a username

Source code in cocorum/basehandles.py

class BaseUserBadge:
    """A badge on a username"""

    def __eq__(self, other: Any) -> bool:
        """Check if this badge is equal to another.

        Args:
            other (Any): Object to compare to.

        Returns:
            Comparison (bool): 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

        return False

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

    def __repr__(self) -> str:
        """String to represent this object"""
        return f"{type(self).__name__}(slug='{self.slug}')"

    @property
    def icon(self) -> bytes:
        """The badge's icon as a bytestring"""
        if not self.__icon:  # We never queried the icon before
            # TODO make the timeout configurable
            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

__eq__(other)

Check if this badge is equal to another.

Parameters:

Name	Type	Description	Default
other	Any	Object to compare to.	required

Returns:

Name	Type	Description
Comparison	bool	Did it fit the criteria?

Source code in cocorum/basehandles.py

def __eq__(self, other: Any) -> bool:
    """Check if this badge is equal to another.

    Args:
        other (Any): Object to compare to.

    Returns:
        Comparison (bool): 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

    return False

__repr__()

String to represent this object

Source code in cocorum/basehandles.py

def __repr__(self) -> str:
    """String to represent this object"""
    return f"{type(self).__name__}(slug='{self.slug}')"

__str__()

The chat user badge in string form

Source code in cocorum/basehandles.py

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

Channel

Bases: Chatter

A channel in the SSE chat

Source code in cocorum/chatapi.py

class Channel(Chatter):
    """A channel in the SSE chat"""

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

    Args:
        jsondata (dict): The JSON data block for the channel
        chat (ChatAPI): Our parent chat API wrapper
        """

        super().__init__(jsondata, chat)

        # Find the user who has this channel
        self.user: Optional[User] = None
        """The user who owns this channel, if we can find them from the chat's user dict"""

        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) -> bool:
        """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) -> int:
        """The ID of this channel in base 10"""
        return int(self["id"])

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

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

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

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

    @property
    def user_id_b10(self) -> int:
        """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 = None instance-attribute

The user who owns this channel, if we can find them from the chat's user dict

user_id property

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

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	Our parent chat API wrapper	required

Source code in cocorum/chatapi.py

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

Args:
    jsondata (dict): The JSON data block for the channel
    chat (ChatAPI): Our parent chat API wrapper
    """

    super().__init__(jsondata, chat)

    # Find the user who has this channel
    self.user: Optional[User] = None
    """The user who owns this channel, if we can find them from the chat's user dict"""

    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

ChatAPI

The Rumble internal chat API

Source code in cocorum/chatapi.py

class ChatAPI():
    """The Rumble internal chat API"""

    def __init__(self, stream_id: SupportsInt | str, servicephp: Optional[ServicePHP] = None, history_len: int = 1000):
        """The Rumble internal chat API

    Args:
        stream_id (int, str): Stream ID in base 10 int or base 36 str.
            WARNING: If a str is passed, this WILL ASSUME BASE 36
            even if only digits are present! Convert to int before passing
            if it is base 10.
        servicephp (ServicePHP): A logged in Service.PHP agent for two-way.
            Defaults to None, view chat as a guest.
        history_len (int): Length of message history to store.
            Defaults to 1000.
            """

        self.stream_id: str = utils.ensure_b36(stream_id)
        """The stream ID in base 36"""

        self.__mailbox = []  # A mailbox if you will
        self.__history = []  # Chat history
        self.history_len: int = history_len
        """How many messages to store in history"""

        self.pinned_message: Optional[Message] = None
        """If a message is pinned, it is assigned to this"""

        self.users: dict[int, User] = {}
        """Users by user ID"""

        self.channels: dict[int, Channel] = {}
        """Channels by channel ID"""

        self.badges: dict[str, UserBadge] = {}
        """User badge types by slug"""

        # Generate our URLs
        self.sse_url: str = static.URI.ChatAPI.sse_stream.format(
            stream_id_b10=self.stream_id_b10)
        """The URL of the chat SSE stream"""

        self.message_api_url = static.URI.ChatAPI.message.format(
            stream_id_b10=self.stream_id_b10)
        """The URL of the chat API"""

        #  Connect to SSE stream
        #  Note: We do NOT want this request to have a timeout
        self.client: ssec.EventSource = ssec.EventSource(
            self.sse_url, headers=static.RequestHeaders.sse_api)
        """The client for the chat SSE stream"""

        self.client.connect()
        self.chat_running: bool = True
        """Status of wether the chat is still open"""

        #  If we have session login, use it
        self.servicephp: ServicePHP = servicephp
        """The ServicePHP wrapper for two-way chat interaction"""

        self.scraper: Optional[scraping.Scraper] = scraping.Scraper(
            self.servicephp) if self.servicephp else None
        """An HTML scraper for getting data like mute records"""

        #  Parse the init data for the stream (must do AFTER we have servicephp)
        self.parse_init_data(self.__next_event_json())

        self.last_send_time: float = 0
        """The last time we sent a message"""

    def __repr__(self) -> str:
        """String to represent this object"""
        return f"{type(self).__name__}(stream_id={self.stream_id})"

    def close(self):
        """Close the chat connection"""
        self.client.close()
        self.chat_running = False

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

    @property
    def history(self) -> tuple[Message, ...]:
        """The chat history, trimmed to history_len"""
        return tuple(self.__history)

    def send_message(self, text: str, channel_id: Optional[SupportsInt] = None) -> (int, User):
        """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 (User): 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": int(channel_id) if channel_id else None,
                }
            },
            #  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"]), User(r.json()["data"]["user"], self)

    def command(self, command_message: str) -> dict:
        """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: SupportsInt) -> bool:
        """Delete a message in chat.

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

    Returns:
        success (bool): Wether the operation succeeded or not.
            NOTE: Method will also print an error message if it failed.
        """

        assert not hasattr(
            message, "deleted") or not message.deleted, "Message was already deleted"

        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

        if hasattr(message, "deleted"):
            message.deleted = True

        return True

    def pin_message(self, message: SupportsInt):
        """Pin a message

        Args:
            message (SupportsInt): Converting this to int must return a chat message ID.
        """

        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: Optional[SupportsInt] = None):
        """Unpin the pinned message

        Args:
            message (SupportsInt): Message to unpin.
                Defaults to None, unpin known 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: str, 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: str):
        """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"
        self.servicephp.unmute_user(record_id)

    def __next_event_json(self) -> dict:
        """Wait for the next event from the SSE and parse the JSON

        Returns:
            json (dict): The next event
        """

        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.client, None)

        # TODO: Will this ever happen?
        except requests.exceptions.ReadTimeout:
            print("Request read timeout.")
            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: dict):
        """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: dict):
        """Parse chat messages from an SSE data JSON

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

        # Add new messages
        self.__mailbox += [Message(message_json, self) for message_json in jsondata["data"].get(
            "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: dict):
        """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"].get("users", []):
            try:
                # Update an existing user's JSON
                self.users[int(user_json["id"])]._jsondata = user_json
            except KeyError:  # User is new
                self.users[int(user_json["id"])] = User(user_json, self)

    def update_channels(self, jsondata: dict):
        """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"].get("channels", []):
            try:
                # Update an existing channel's JSON
                self.channels[int(channel_json["id"])]._jsondata = channel_json
            except KeyError:  # Channel is new
                self.channels.update(
                    {int(channel_json["id"]): Channel(channel_json, self)})

    def load_badges(self, jsondata: dict):
        """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: UserBadge(
            badge_slug, jsondata["data"]["config"]["badges"][badge_slug], self) for badge_slug in jsondata["data"]["config"]["badges"].keys()}

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

    def get_message(self) -> Message | None:
        """Return the next chat message (parsing any additional data).
        Waits for it to come in, returns None if chat closed.

        Returns:
            result (Message | None): Either the next chat message or NoneType.
        """

        # 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 = Message(
                    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

badges = {} instance-attribute

User badge types by slug

channels = {} instance-attribute

Channels by channel ID

chat_running = True instance-attribute

Status of wether the chat is still open

client = ssec.EventSource(self.sse_url, headers=(static.RequestHeaders.sse_api)) instance-attribute

The client for the chat SSE stream

history property

The chat history, trimmed to history_len

history_len = history_len instance-attribute

How many messages to store in history

last_send_time = 0 instance-attribute

The last time we sent a message

message_api_url = static.URI.ChatAPI.message.format(stream_id_b10=(self.stream_id_b10)) instance-attribute

The URL of the chat API

pinned_message = None instance-attribute

If a message is pinned, it is assigned to this

scraper = scraping.Scraper(self.servicephp) if self.servicephp else None instance-attribute

An HTML scraper for getting data like mute records

servicephp = servicephp instance-attribute

The ServicePHP wrapper for two-way chat interaction

session_cookie property

The session cookie we are logged in with

sse_url = static.URI.ChatAPI.sse_stream.format(stream_id_b10=(self.stream_id_b10)) instance-attribute

The URL of the chat SSE stream

stream_id = utils.ensure_b36(stream_id) instance-attribute

The stream ID in base 36

stream_id_b10 property

The chat ID in use

users = {} instance-attribute

Users by user ID

__init__(stream_id, servicephp=None, history_len=1000)

The Rumble internal chat API

Parameters:

Name	Type	Description	Default
stream_id	(int, str)	Stream ID in base 10 int or base 36 str. WARNING: If a str is passed, this WILL ASSUME BASE 36 even if only digits are present! Convert to int before passing if it is base 10.	required
servicephp	ServicePHP	A logged in Service.PHP agent for two-way. Defaults to None, view chat as a guest.	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: SupportsInt | str, servicephp: Optional[ServicePHP] = None, history_len: int = 1000):
    """The Rumble internal chat API

Args:
    stream_id (int, str): Stream ID in base 10 int or base 36 str.
        WARNING: If a str is passed, this WILL ASSUME BASE 36
        even if only digits are present! Convert to int before passing
        if it is base 10.
    servicephp (ServicePHP): A logged in Service.PHP agent for two-way.
        Defaults to None, view chat as a guest.
    history_len (int): Length of message history to store.
        Defaults to 1000.
        """

    self.stream_id: str = utils.ensure_b36(stream_id)
    """The stream ID in base 36"""

    self.__mailbox = []  # A mailbox if you will
    self.__history = []  # Chat history
    self.history_len: int = history_len
    """How many messages to store in history"""

    self.pinned_message: Optional[Message] = None
    """If a message is pinned, it is assigned to this"""

    self.users: dict[int, User] = {}
    """Users by user ID"""

    self.channels: dict[int, Channel] = {}
    """Channels by channel ID"""

    self.badges: dict[str, UserBadge] = {}
    """User badge types by slug"""

    # Generate our URLs
    self.sse_url: str = static.URI.ChatAPI.sse_stream.format(
        stream_id_b10=self.stream_id_b10)
    """The URL of the chat SSE stream"""

    self.message_api_url = static.URI.ChatAPI.message.format(
        stream_id_b10=self.stream_id_b10)
    """The URL of the chat API"""

    #  Connect to SSE stream
    #  Note: We do NOT want this request to have a timeout
    self.client: ssec.EventSource = ssec.EventSource(
        self.sse_url, headers=static.RequestHeaders.sse_api)
    """The client for the chat SSE stream"""

    self.client.connect()
    self.chat_running: bool = True
    """Status of wether the chat is still open"""

    #  If we have session login, use it
    self.servicephp: ServicePHP = servicephp
    """The ServicePHP wrapper for two-way chat interaction"""

    self.scraper: Optional[scraping.Scraper] = scraping.Scraper(
        self.servicephp) if self.servicephp else None
    """An HTML scraper for getting data like mute records"""

    #  Parse the init data for the stream (must do AFTER we have servicephp)
    self.parse_init_data(self.__next_event_json())

    self.last_send_time: float = 0
    """The last time we sent a message"""

__next_event_json()

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

Returns:

Name	Type	Description
json	dict	The next event

Source code in cocorum/chatapi.py

def __next_event_json(self) -> dict:
    """Wait for the next event from the SSE and parse the JSON

    Returns:
        json (dict): The next event
    """

    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.client, None)

    # TODO: Will this ever happen?
    except requests.exceptions.ReadTimeout:
        print("Request read timeout.")
        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)

__repr__()

String to represent this object

Source code in cocorum/chatapi.py

def __repr__(self) -> str:
    """String to represent this object"""
    return f"{type(self).__name__}(stream_id={self.stream_id})"

clear_mailbox()

Delete anything in the mailbox

Source code in cocorum/chatapi.py

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

close()

Close the chat connection

Source code in cocorum/chatapi.py

def close(self):
    """Close the chat connection"""
    self.client.close()
    self.chat_running = False

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) -> dict:
    """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	SupportsInt	Object which when converted to integer is the target message ID.	required

Returns:

Name	Type	Description
success	bool	Wether the operation succeeded or not. NOTE: Method will also print an error message if it failed.

Source code in cocorum/chatapi.py

def delete_message(self, message: SupportsInt) -> bool:
    """Delete a message in chat.

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

Returns:
    success (bool): Wether the operation succeeded or not.
        NOTE: Method will also print an error message if it failed.
    """

    assert not hasattr(
        message, "deleted") or not message.deleted, "Message was already deleted"

    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

    if hasattr(message, "deleted"):
        message.deleted = True

    return True

get_message()

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

Returns:

Name	Type	Description
result	Message | None	Either the next chat message or NoneType.

Source code in cocorum/chatapi.py

def get_message(self) -> Message | None:
    """Return the next chat message (parsing any additional data).
    Waits for it to come in, returns None if chat closed.

    Returns:
        result (Message | None): Either the next chat message or NoneType.
    """

    # 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 = Message(
                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: dict):
    """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: UserBadge(
        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: str, 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: dict):
    """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

Parameters:

Name	Type	Description	Default
message	SupportsInt	Converting this to int must return a chat message ID.	required

Source code in cocorum/chatapi.py

def pin_message(self, message: SupportsInt):
    """Pin a message

    Args:
        message (SupportsInt): Converting this to int must return a chat message ID.
    """

    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	User	Your current chat user information.

Source code in cocorum/chatapi.py

def send_message(self, text: str, channel_id: Optional[SupportsInt] = None) -> (int, User):
    """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 (User): 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": int(channel_id) if channel_id else None,
            }
        },
        #  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"]), User(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: str):
    """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"
    self.servicephp.unmute_user(record_id)

unpin_message(message=None)

Unpin the pinned message

Parameters:

Name	Type	Description	Default
message	SupportsInt	Message to unpin. Defaults to None, unpin known pinned message.	None

Source code in cocorum/chatapi.py

def unpin_message(self, message: Optional[SupportsInt] = None):
    """Unpin the pinned message

    Args:
        message (SupportsInt): Message to unpin.
            Defaults to None, unpin known 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: dict):
    """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"].get("channels", []):
        try:
            # Update an existing channel's JSON
            self.channels[int(channel_json["id"])]._jsondata = channel_json
        except KeyError:  # Channel is new
            self.channels.update(
                {int(channel_json["id"]): Channel(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: dict):
    """Parse chat messages from an SSE data JSON

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

    # Add new messages
    self.__mailbox += [Message(message_json, self) for message_json in jsondata["data"].get(
        "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: dict):
    """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"].get("users", []):
        try:
            # Update an existing user's JSON
            self.users[int(user_json["id"])]._jsondata = user_json
        except KeyError:  # User is new
            self.users[int(user_json["id"])] = User(user_json, self)

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: dict, chat: ChatAPI):
        """Object in the internal chat API

    Args:
        jsondata (dict): The JSON data block for the object
        chat (ChatAPI): Our parent chat API wrapper
        """

        JSONObj.__init__(self, jsondata)
        self.chat: ChatAPI = chat
        """Our parent chat API wrapper"""

chat = chat instance-attribute

Our parent chat API wrapper

__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	Our parent chat API wrapper	required

Source code in cocorum/chatapi.py

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

Args:
    jsondata (dict): The JSON data block for the object
    chat (ChatAPI): Our parent chat API wrapper
    """

    JSONObj.__init__(self, jsondata)
    self.chat: ChatAPI = chat
    """Our parent chat API wrapper"""

Chatter

Bases: JSONUserAction, ChatAPIObj

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

Source code in cocorum/chatapi.py

class Chatter(JSONUserAction, ChatAPIObj):
    """A user or channel in the internal chat API (abstract)"""

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

        Args:
            jsondata (dict): The JSON data block for the user/channel
            chat (ChatAPI): Our parent chat API wrapper
        """
        ChatAPIObj.__init__(self, jsondata, chat)
        JSONUserAction.__init__(self, jsondata)

    def __repr__(self) -> str:
        """String to represent this object"""
        return f"{type(self).__name__}(username=\"{self.username}\", user_id={self.user_id}, channel_id={self.channel_id})"

    @property
    def link(self) -> str:
        """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	Our parent chat API wrapper	required

Source code in cocorum/chatapi.py

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

    Args:
        jsondata (dict): The JSON data block for the user/channel
        chat (ChatAPI): Our parent chat API wrapper
    """
    ChatAPIObj.__init__(self, jsondata, chat)
    JSONUserAction.__init__(self, jsondata)

__repr__()

String to represent this object

Source code in cocorum/chatapi.py

def __repr__(self) -> str:
    """String to represent this object"""
    return f"{type(self).__name__}(username=\"{self.username}\", user_id={self.user_id}, channel_id={self.channel_id})"

GiftPurchaseNotification

Bases: ChatAPIObj

A subscription gift under a message

Source code in cocorum/chatapi.py

class GiftPurchaseNotification(ChatAPIObj):
    """A subscription gift under a message"""

    def __init__(self, jsondata: str, message: Message):
        """A subscription gift under a message

    Args:
        jsondata (dict): The JSON data block for the message.
        message (Message): Our parent ChatAPI message
        """

        super().__init__(jsondata, message.chat)
        self.message: Message = message
        """Our parent ChatAPI message"""

    def __repr__(self) -> str:
        """String to represent this object"""
        return f"{type(self).__name__}(purchased_by='{self.purchased_by}', <for ${self.message.amount_cents / 100:.02f}>)"

    @property
    def total_gifts(self) -> int:
        """The number of subscriptions in this gift"""
        return self["total_gifts"]

    @property
    def gift_type(self) -> str:
        """TODO"""
        return self["gift_type"]

    @property
    def video_id(self) -> int:
        """The numeric ID of the stream this gift was sent on, in base 10"""
        return self.chat.stream_id_b10

    @property
    def video_id_b10(self) -> int:
        """The numeric ID of the stream this gift was sent on, in base 10"""
        return self.video_id

    @property
    def video_id_b36(self) -> str:
        """The numeric ID of the stream this gift was sent on, in base 36"""
        return utils.base_10_to_36(self.video_id)

    @property
    def purchased_by(self) -> str:
        """The username of who purchased this gift"""
        return self.message.user.username

    @property
    def creator_user_id(self) -> int:
        """The numeric ID of the user whose stream this gift was given on, in base 10"""
        return self["creator_user_id"]

    @property
    def creator_user_id_b10(self) -> int:
        """The numeric ID of the user whose stream this gift was given on, in base 10"""
        return self.creator_user_id

    @property
    def creator_user_id_b36(self) -> str:
        """The numeric ID of the user whose stream this gift was given on, in base 36"""
        return utils.base_10_to_36(self.creator_user_id)

    @property
    def creator_channel_id(self) -> int:
        """The numeric ID of the channel whose stream this gift was given on, in base 10 (can be zero)"""
        return self["creator_channel_id"] if self["creator_channel_id"] else 0

    @property
    def creator_channel_id_b10(self) -> int:
        """The numeric ID of the channel whose stream this gift was given on, in base 10 (can be zero)"""
        return self.creator_channel_id

    @property
    def creator_channel_id_b36(self) -> str:
        """The numeric ID of the channel whose stream this gift was given on, in base 36 (can be zero)"""
        return utils.base_10_to_36(self.creator_channel_id)

creator_channel_id property

The numeric ID of the channel whose stream this gift was given on, in base 10 (can be zero)

creator_channel_id_b10 property

The numeric ID of the channel whose stream this gift was given on, in base 10 (can be zero)

creator_channel_id_b36 property

The numeric ID of the channel whose stream this gift was given on, in base 36 (can be zero)

creator_user_id property

The numeric ID of the user whose stream this gift was given on, in base 10

creator_user_id_b10 property

The numeric ID of the user whose stream this gift was given on, in base 10

creator_user_id_b36 property

The numeric ID of the user whose stream this gift was given on, in base 36

gift_type property

TODO

message = message instance-attribute

Our parent ChatAPI message

purchased_by property

The username of who purchased this gift

total_gifts property

The number of subscriptions in this gift

video_id property

The numeric ID of the stream this gift was sent on, in base 10

video_id_b10 property

The numeric ID of the stream this gift was sent on, in base 10

video_id_b36 property

The numeric ID of the stream this gift was sent on, in base 36

__init__(jsondata, message)

A subscription gift under a message

Parameters:

Name	Type	Description	Default
jsondata	dict	The JSON data block for the message.	required
message	Message	Our parent ChatAPI message	required

Source code in cocorum/chatapi.py

def __init__(self, jsondata: str, message: Message):
    """A subscription gift under a message

Args:
    jsondata (dict): The JSON data block for the message.
    message (Message): Our parent ChatAPI message
    """

    super().__init__(jsondata, message.chat)
    self.message: Message = message
    """Our parent ChatAPI message"""

__repr__()

String to represent this object

Source code in cocorum/chatapi.py

def __repr__(self) -> str:
    """String to represent this object"""
    return f"{type(self).__name__}(purchased_by='{self.purchased_by}', <for ${self.message.amount_cents / 100:.02f}>)"

Message

Bases: ChatAPIObj

A single chat message in the internal chat API

Source code in cocorum/chatapi.py

class Message(ChatAPIObj):
    """A single chat message in the internal chat API"""

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

    Args:
        jsondata (dict): The JSON data block for the message
        chat (ChatAPI): Our parent chat API wrapper
        """

        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

        self.deleted: bool = False
        """Was this message deleted?"""

    def __eq__(self, other: Any) -> bool:
        """Compare this chat message with another

        Args:
            other (Any): Object to compare to.

        Returns:
            Comparison (bool): 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

        return False

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

    def __repr__(self) -> str:
        """String to represent this object"""
        return f"{type(self).__name__}(<from '{self.user.username}'>, text=\"{self.text}\")"

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

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

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

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

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

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

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

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

    @property
    def channel_id(self) -> int | None:
        """The numeric ID of the channel who posted the message, if there is one, in base 10"""
        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) -> int | None:
        """The ID of the channel who posted the message, if there is one, in base 10"""
        return self.channel_id

    @property
    def channel_id_b36(self) -> str | None:
        """The ID of the channel who posted the message, if there is one, in base 36"""
        return utils.base_10_to_36(self.channel_id) if self.channel_id else None

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

    @property
    def user(self) -> User | None:
        """Reference to the user who posted this message, if we can find them"""
        return self.chat.users.get(self.user_id)

    @property
    def channel(self) -> Channel | None:
        """Reference to the channel that posted this message, if there was one and we can find it"""
        if not self.channel_id:
            return None

        return self.chat.channels.get(self.channel_id)

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

    @property
    def rant_price_cents(self) -> int:
        """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) -> float:
        """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) -> float:
        """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) -> dict | False:
        """Are we a raid notification? Returns associated JSON data if yes, False if no"""
        return self.get("raid_notification", False)

    @property
    def gift_purchase_notification(self) -> GiftPurchaseNotification | False:
        """Are we a gifted subs notification? Returns JSON wrapper if yes, False if no

    Returns:
        Data (GiftPurchaseNotification | bool): A simple container for the data, or False"""

        if self.get("gift_purchase_notification"):
            return GiftPurchaseNotification(self["gift_purchase_notification"], self)

        return False

    def delete(self) -> bool:
        """Delete this message from the chat"""
        return self.chat.delete(self)

    def pin(self):
        """Pin this message"""
        return self.chat.pin_message(self)

    def unpin(self):
        """Unpin this message if it was pinned"""
        return self.chat.unpin_message(self)

channel property

Reference to the channel that posted this message, if there was one and we can find it

channel_id property

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

channel_id_b10 property

The ID of the channel who posted the message, if there is one, in base 10

channel_id_b36 property

The ID of the channel who posted the message, if there is one, in base 36

deleted = False instance-attribute

Was this message deleted?

gift_purchase_notification property

Are we a gifted subs notification? Returns JSON wrapper if yes, False if no

Returns:

Name	Type	Description
Data	GiftPurchaseNotification | bool	A simple container for the data, or False

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, if we can find them

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	Any	Object to compare to.	required

Returns:

Name	Type	Description
Comparison	bool	Did it fit the criteria?

Source code in cocorum/chatapi.py

def __eq__(self, other: Any) -> bool:
    """Compare this chat message with another

    Args:
        other (Any): Object to compare to.

    Returns:
        Comparison (bool): 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

    return False

__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	Our parent chat API wrapper	required

Source code in cocorum/chatapi.py

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

Args:
    jsondata (dict): The JSON data block for the message
    chat (ChatAPI): Our parent chat API wrapper
    """

    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

    self.deleted: bool = False
    """Was this message deleted?"""

__int__()

The chat message in integer (ID) form

Source code in cocorum/chatapi.py

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

__repr__()

String to represent this object

Source code in cocorum/chatapi.py

def __repr__(self) -> str:
    """String to represent this object"""
    return f"{type(self).__name__}(<from '{self.user.username}'>, text=\"{self.text}\")"

__str__()

The chat message in string form

Source code in cocorum/chatapi.py

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

delete()

Delete this message from the chat

Source code in cocorum/chatapi.py

def delete(self) -> bool:
    """Delete this message from the chat"""
    return self.chat.delete(self)

pin()

Pin this message

Source code in cocorum/chatapi.py

def pin(self):
    """Pin this message"""
    return self.chat.pin_message(self)

unpin()

Unpin this message if it was pinned

Source code in cocorum/chatapi.py

def unpin(self):
    """Unpin this message if it was pinned"""
    return self.chat.unpin_message(self)

User

Bases: Chatter, BaseUser

User in the internal chat API

Source code in cocorum/chatapi.py

class User(Chatter, BaseUser):
    """User in the internal chat API"""

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

    Args:
        jsondata (dict): The JSON data block for the user
        chat (ChatAPI): Our parent chat API wrapper
        """

        Chatter.__init__(self, jsondata, chat)
        self.previous_channel_ids: list[int] = []
        """A list of channels the user has appeared as, including the current one"""

        self._set_channel_id: int = None
        """Channel ID set from message"""

        self.servicephp: Optional[ServicePHP] = self.chat.servicephp
        """Our parent chat API's ServicePHP instance, if it has one"""

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

    @property
    def channel_id(self) -> int:
        """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) -> int:
        """The numeric channel ID that the user is appearing with in base 10"""
        return self.channel_id

    @property
    def channel_id_b36(self) -> str:
        """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) -> bool:
        """Is this user following the livestreaming channel?"""
        return self["is_follower"]

    @property
    def color(self) -> tuple[int, int, int]:
        """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) -> list[str]:
        """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?

previous_channel_ids = [] instance-attribute

A list of channels the user has appeared as, including the current one

servicephp = self.chat.servicephp instance-attribute

Our parent chat API's ServicePHP instance, if it has one

user_id property

The numeric ID of the user in base 10

__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	Our parent chat API wrapper	required

Source code in cocorum/chatapi.py

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

Args:
    jsondata (dict): The JSON data block for the user
    chat (ChatAPI): Our parent chat API wrapper
    """

    Chatter.__init__(self, jsondata, chat)
    self.previous_channel_ids: list[int] = []
    """A list of channels the user has appeared as, including the current one"""

    self._set_channel_id: int = None
    """Channel ID set from message"""

    self.servicephp: Optional[ServicePHP] = self.chat.servicephp
    """Our parent chat API's ServicePHP instance, if it has one"""

UserBadge

Bases: ChatAPIObj, BaseUserBadge

A badge of a user

Source code in cocorum/chatapi.py

class UserBadge(ChatAPIObj, BaseUserBadge):
    """A badge of a user"""

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

    Args:
        slug (str): The unique identification for this badge
        jsondata (dict): The JSON data block for the user badge
        chat (ChatAPI): Our parent chat API wrapper
        """

        ChatAPIObj.__init__(self, jsondata, chat)
        self.slug: str = slug
        """The unique identification for this badge"""

        self.__icon = None

    @property
    def label(self) -> dict[str, str]:
        """A dictionary of lang:label pairs"""
        return self["label"]

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

icon_url property

The URL of the badge's icon

label property

A dictionary of lang:label pairs

slug = slug instance-attribute

The unique identification for this badge

__init__(slug, jsondata, chat)

A user badge in the internal chat API

Parameters:

Name	Type	Description	Default
slug	str	The unique identification for this badge	required
jsondata	dict	The JSON data block for the user badge	required
chat	ChatAPI	Our parent chat API wrapper	required

Source code in cocorum/chatapi.py

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

Args:
    slug (str): The unique identification for this badge
    jsondata (dict): The JSON data block for the user badge
    chat (ChatAPI): Our parent chat API wrapper
    """

    ChatAPIObj.__init__(self, jsondata, chat)
    self.slug: str = slug
    """The unique identification for this badge"""

    self.__icon = None

S.D.G.