cocorum.servicephp

The primary use from this module is the ServicePHP class, a wrapper for Rumble's internal service.php API. Login is performed via this API, as well as most Rumble functions other than chat messaging and video upload. All other classes are supporting sub-classes.

Service.PHP interactions

Control Rumble via Service.PHP

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.

APIComment

Bases: JSONObj, BaseComment

A comment on a video as returned by a successful attempt to make it

Source code in cocorum/servicephp.py

class APIComment(JSONObj, BaseComment):
    """A comment on a video as returned by a successful attempt to make it"""

    def __init__(self, jsondata: dict, servicephp: ServicePHP):
        """A comment on a video as returned by a successful attempt to make it

        Args:
            jsondata (dict): The JSON block for a single comment.
            servicephp (ServicePHP): The ServicePHP object that spawned us.
        """

        JSONObj.__init__(self, jsondata)
        self.servicephp: ServicePHP = servicephp
        """Our parent ServicePHP wrapper"""

        # Badges of the user who commented if we have them
        if self.get("comment_user_badges"):
            self.user_badges = {
                slug: APIUserBadge(slug, data)
                for slug, data in self["comment_user_badges"].items()
            }

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

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

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

    @property
    def user_display(self) -> str:
        """The display name of the user who commented"""
        return self["comment_user_display"]

    @property
    def tree_size(self):
        """TODO"""
        return self["comment_tree_size"]

comment_id property

The numeric ID of the comment in base 10

servicephp = servicephp instance-attribute

Our parent ServicePHP wrapper

text property

The text of the comment

tree_size property

TODO

user_display property

The display name of the user who commented

__init__(jsondata, servicephp)

A comment on a video as returned by a successful attempt to make it

Parameters:

Name	Type	Description	Default
jsondata	dict	The JSON block for a single comment.	required
servicephp	ServicePHP	The ServicePHP object that spawned us.	required

Source code in cocorum/servicephp.py

def __init__(self, jsondata: dict, servicephp: ServicePHP):
    """A comment on a video as returned by a successful attempt to make it

    Args:
        jsondata (dict): The JSON block for a single comment.
        servicephp (ServicePHP): The ServicePHP object that spawned us.
    """

    JSONObj.__init__(self, jsondata)
    self.servicephp: ServicePHP = servicephp
    """Our parent ServicePHP wrapper"""

    # Badges of the user who commented if we have them
    if self.get("comment_user_badges"):
        self.user_badges = {
            slug: APIUserBadge(slug, data)
            for slug, data in self["comment_user_badges"].items()
        }

__repr__()

String to represent this object

Source code in cocorum/servicephp.py

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

APIPlaylist

Bases: JSONObj, BasePlaylist

Playlist as returned by the API

Source code in cocorum/servicephp.py

class APIPlaylist(JSONObj, BasePlaylist):
    """Playlist as returned by the API"""

    def __init__(self, jsondata: dict, servicephp: ServicePHP):
        """Playlist as returned by the API.

        Args:
            jsondata (dict): The JSON data block of a playlist.
            servicephp (ServicePHP): The ServicePHP object that spawned us.
        """

        JSONObj.__init__(self, jsondata)
        self.servicephp: ServicePHP = servicephp
        """Our parent ServicePHP wrapper"""

        self.user: APIUser = APIUser(jsondata["user"], self.servicephp)
        """The user that created this playlist"""

    @property
    def playlist_id(self) -> str:
        """The numeric playlist ID in base 64"""
        return self["id"]

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

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

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

    @property
    def url(self) -> str:
        """The URL of the playlist"""
        return self["url"]

    @property
    def channel(self):
        """The channel the playlist is under, can be None"""
        # TODO Unknown return type
        return self["channel"]

    @property
    def created_on(self) -> float:
        """The time the playlist was created in seconds since epoch"""
        return utils.parse_timestamp(self["created_on"])

    @property
    def updated_on(self) -> float:
        """The time the playlist was last updated in seconds since epoch"""
        return utils.parse_timestamp(self["updated_on"])

    @property
    def permissions(self):
        """The permissions the ServicePHP user has on this playlist"""
        # TODO Unknown return type
        return self["permissions"]

    @property
    def num_items(self) -> int:
        """The number of items in the playlist"""
        return self["num_items"]

    @property
    def is_following(self) -> bool:
        """TODO -> Bool"""
        return self["is_following"]

    @property
    def items(self):
        """The items of the playlist"""
        # TODO unwrapped JSON data I think
        return self["items"]

    @property
    def extra(self):
        """TODO -> None, unknown"""
        return self["extra"]

channel property

The channel the playlist is under, can be None

created_on property

The time the playlist was created in seconds since epoch

description property

The description of the playlist

extra property

TODO -> None, unknown

is_following property

TODO -> Bool

items property

The items of the playlist

num_items property

The number of items in the playlist

permissions property

The permissions the ServicePHP user has on this playlist

playlist_id property

The numeric playlist ID in base 64

servicephp = servicephp instance-attribute

Our parent ServicePHP wrapper

title property

The title of the playlist

updated_on property

The time the playlist was last updated in seconds since epoch

url property

The URL of the playlist

user = APIUser(jsondata['user'], self.servicephp) instance-attribute

The user that created this playlist

visibility property

The visibility of the playlist

__init__(jsondata, servicephp)

Playlist as returned by the API.

Parameters:

Name	Type	Description	Default
jsondata	dict	The JSON data block of a playlist.	required
servicephp	ServicePHP	The ServicePHP object that spawned us.	required

Source code in cocorum/servicephp.py

def __init__(self, jsondata: dict, servicephp: ServicePHP):
    """Playlist as returned by the API.

    Args:
        jsondata (dict): The JSON data block of a playlist.
        servicephp (ServicePHP): The ServicePHP object that spawned us.
    """

    JSONObj.__init__(self, jsondata)
    self.servicephp: ServicePHP = servicephp
    """Our parent ServicePHP wrapper"""

    self.user: APIUser = APIUser(jsondata["user"], self.servicephp)
    """The user that created this playlist"""

APIUser

Bases: JSONObj, BaseUser

User data as returned by the API

Source code in cocorum/servicephp.py

class APIUser(JSONObj, BaseUser):
    """User data as returned by the API"""

    def __init__(self, jsondata: dict, servicephp: ServicePHP):
        """User data as returned by the API.

        Args:
            jsondata (dict): The JSON data block of a single user.
            servicephp (ServicePHP): The ServicePHP object that spawned us.
        """

        JSONObj.__init__(self, jsondata)
        self.servicephp: ServicePHP = servicephp
        """Our parent ServicePHP wrapper"""

        # Our profile picture data
        self.__picture = None

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

    @property
    def user_id(self) -> int:
        """The numeric ID of the user in base 10"""
        return self["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 username(self) -> str:
        """The username of the user"""
        return self["username"]

    @property
    def picture_url(self) -> str:
        """The URL of the user's profile picture"""
        return self["picture"]

    @property
    def picture(self) -> bytes:
        """The user's profile picture as a bytes string"""
        if not self.picture_url:  # The profile picture is blank
            return b""

        if not self.__picture:  # We never queried the profile pic before
            response = requests.get(
                self.picture_url, timeout=static.Delays.request_timeout
            )
            assert response.status_code == 200, "Status code " + str(
                response.status_code
            )

            self.__picture = response.content

        return self.__picture

    @property
    def verified_badge(self):
        """Is the user verified?"""
        # TODO Unknown return type
        return self["verified_badge"]

    @property
    def followers(self) -> int:
        """The number of followers this user has"""
        return self["followers"]

    @property
    def followed(self) -> bool:
        """TODO -> Bool"""
        return self["followed"]

followed property

TODO -> Bool

followers property

The number of followers this user has

picture property

The user's profile picture as a bytes string

picture_url property

The URL of the user's profile picture

servicephp = servicephp instance-attribute

Our parent ServicePHP wrapper

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

username property

The username of the user

verified_badge property

Is the user verified?

__init__(jsondata, servicephp)

User data as returned by the API.

Parameters:

Name	Type	Description	Default
jsondata	dict	The JSON data block of a single user.	required
servicephp	ServicePHP	The ServicePHP object that spawned us.	required

Source code in cocorum/servicephp.py

def __init__(self, jsondata: dict, servicephp: ServicePHP):
    """User data as returned by the API.

    Args:
        jsondata (dict): The JSON data block of a single user.
        servicephp (ServicePHP): The ServicePHP object that spawned us.
    """

    JSONObj.__init__(self, jsondata)
    self.servicephp: ServicePHP = servicephp
    """Our parent ServicePHP wrapper"""

    # Our profile picture data
    self.__picture = None

__repr__()

String to represent this object

Source code in cocorum/servicephp.py

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

APIUserBadge

Bases: JSONObj, BaseUserBadge

A badge of a user as returned by the API

Source code in cocorum/servicephp.py

class APIUserBadge(JSONObj, BaseUserBadge):
    """A badge of a user as returned by the API"""

    def __init__(self, slug: str, jsondata: dict):
        """A badge of a user as returned by the API.

        Args:
            slug (str): The string identifier of the badge.
            jsondata (dict): The JSON data block of the badge.
        """

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

        self.__icon = None

    @property
    def label(self) -> 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 type

__init__(slug, jsondata)

A badge of a user as returned by the API.

Parameters:

Name	Type	Description	Default
slug	str	The string identifier of the badge.	required
jsondata	dict	The JSON data block of the badge.	required

Source code in cocorum/servicephp.py

def __init__(self, slug: str, jsondata: dict):
    """A badge of a user as returned by the API.

    Args:
        slug (str): The string identifier of the badge.
        jsondata (dict): The JSON data block of the badge.
    """

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

    self.__icon = None

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

ServicePHP

Interact with Rumble's service.php API

Source code in cocorum/servicephp.py

class ServicePHP:
    """Interact with Rumble's service.php API"""

    def __init__(self, username: str = None, session: str | dict = None):
        """Interact with Rumble's service.php API.

        Args:
            username (str): The username we will be working under.
                Defaults to None, refer to session.
            session (str | dict): A pre-existing session token.
                Defaults to None, await login.
        """

        assert username or session, "Must pass at least one thing to log in with"

        # Set up initial auth data variables
        self.username: str | None = username
        """The username this ServicePHP wrapper is working under"""

        self.session_cookie: dict | None = None
        """The session token cookie we received on login. Can be reused if not logged out."""

        # Session is the token directly
        if isinstance(session, str):
            self.session_cookie = {static.Misc.session_token_key: session}

        # Session is a cookie dict
        elif isinstance(session, dict):
            assert session.get(
                static.Misc.session_token_key
            ), f"Session cookie dict must have '{static.Misc.session_token_key}' as key."
            self.session_cookie = session

        # Session was passed but it is not anything we can use
        elif session is not None:
            raise ValueError(
                f"Session must be a token str or cookie dict, got {type(session)}"
            )

        # If a session cookie was passed, test it
        assert not self.session_cookie or utils.test_session_cookie(
            self.session_cookie
        ), "Session cookie is invalid."

        # Stored ID of the logged in user
        self.__user_id = None

    def __repr__(self) -> str:
        """String to represent this object"""
        return f"{type(self).__name__}(username={self.username}, <{'not ' * bool(self.session_cookie)}logged in>)"

    @property
    def session_token(self) -> str:
        """The token stored in our session cookie, empty if if we are not logged in"""
        return (
            self.session_cookie[static.Misc.session_token_key]
            if self.session_cookie
            else ""
        )

    @property
    def user_id(self) -> int:
        """The numeric ID of the logged in user in base 10"""
        # We do not have a user ID, extract it from the unread notifications response
        if self.__user_id is None:
            j = self.sphp_request(
                "user.has_unread_notifications",
                method="GET",
            ).json()
            self.__user_id = utils.base_36_to_10(
                j["user"]["id"].removeprefix("_"))

        return self.__user_id

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

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

    def sphp_request(
        self,
        service_name: str,
        data: dict = {},
        additional_params: dict = {},
        logged_in: bool = True,
        method: str = "POST",
    ) -> requests.Response:
        """
        Make a request to Service.PHP with common settings

        Args:
            service_name (str): The name parameter of the specific PHP service
            data (dict): Form data.
                Defaults to {}.
            additional_params (dict): Any additional query string parameters.
                Defaults to {}.
            logged_in (bool): Should the request use the session cookie?
                Defaults to True.
            method (str): What request method to use.
                Defaults to 'POST'.

        Returns:
            response (requests.Response): The response from Rumble.
        """
        assert self.session_cookie or not logged_in, "Not logged in yet."

        params = {"name": service_name}
        params.update(additional_params)
        r = requests.request(
            method,
            static.URI.servicephp,
            params=params,
            data=data,
            headers=static.RequestHeaders.user_agent,
            cookies=self.session_cookie if logged_in else None,
            timeout=static.Delays.request_timeout,
        )
        assert (
            r.status_code == 200
        ), f"Service.PHP request for {service_name} failed: {r}\n{r.text}"
        # If the request json has a data -> success value, make sure it is True
        d = r.json().get("data")
        if isinstance(d, dict):
            assert d.get(
                "success", True
            ), f"Service.PHP request for {service_name} failed: \n{r.text}"
        # Data was not a dict but was not empty
        elif d:
            print(
                f"Service.PHP request for {service_name} did not fail but returned unknown data type {type(d)}: {d}"
            )

        return r

    def get_hashed_password(self, password: str) -> str:
        """Use Rumble's salts to hash a pasword for login

        Args:
            password (str): The password to hash with the salts.

        Returns:
            hashed (str): The hashed password to use in the rest of the login.
        """

        # Get salts
        r = self.sphp_request(
            "user.get_salts",
            data={"username": self.username},
            logged_in=False,
            additional_params={"response_type": "session"},
        )
        salts = r.json()["data"]["salts"]

        return ",".join(utils.calc_password_hashes(password, salts))

    def get_2fa_first_step(self, password_hashes: str) -> TwoFacAuth:
        """Get 2FA info on our account as a first 2FA step

        Args:
            password_hashes (str): The hashed password.

        Returns:
            info (TwoFacAuth): Information on 2FA."""

        r = self.sphp_request(
            "user.2fa.first_step",
            data={
                "legacy_password": 1,  # TODO
                "login": self.username,
                "password": password_hashes,
                "redirect_url": static.URI.rumble_base,
            },
            logged_in=False,
        )

        return TwoFacAuth(r.json()["data"], self)

    def login_basic(self, password: str) -> TwoFacAuth | None:
        """Perform a basic password authentication as the first step of login.
            If successful, sets our session token and returns None.
            If incomplete, returns TwoFacAuth() for next step.

        Args:
            password (str): The password to login with.

        Returns:
            result (TwoFacAuth | None): The result of the first login step.
                None means login is complete.
        """

        # Get the password hash to start off
        ph = self.get_hashed_password(password)

        # Check for 2FA
        two_fac_info = self.get_2fa_first_step(ph)

        # 2FA is enabled
        if two_fac_info.totp_id:  # two_fac_auth.options:
            return two_fac_info

        # Get session token for non-2FA
        r = self.sphp_request(
            "user.login",
            data={
                "username": self.username,
                # Hash the password using the salts
                "password_hashes": ph,
            },
            logged_in=False,
        )

        j = r.json()
        session_token = j["data"]["session"]
        assert session_token, f"Login failed: No token returned\n{r.json()}"

        self.session_cookie = {static.Misc.session_token_key: session_token}
        return None

    def login_second_factor(self, two_fac_auth: TwoFacAuth, code: str):
        """Complete login with a 2FA code.

        Args:
            two_fac_auth (TwoFacAuth): 2FA information on this account.
            code (str): The code to enter."""

        # If this fails, sphp_request will raise AssertionError
        r = self.sphp_request(
            "user.2fa.verify_totp",
            data={
                "code": code,
                "redirect_url": static.URI.rumble_base,
                "totp_id": two_fac_auth.totp_id,
                "user_key": two_fac_auth.user_key,
            },
            logged_in=False,
        )

        cookie_start = static.Misc.session_token_key + "="
        for piece in r.headers["Set-Cookie"].split():
            if piece.startswith(cookie_start):
                self.session_cookie = {
                    static.Misc.session_token_key: piece.removeprefix(
                        cookie_start
                    ).removesuffix(";")
                }
                return
        raise ValueError(
            "Login failed: Did not find session token in 2FA response headers."
        )

    def logout(self):
        """Log out this ServicePHP instance"""
        self.sphp_request("user.logout", method="GET")
        self.session_cookie = None

    def chat_pin(
        self, stream_id: SupportsInt | str, message: SupportsInt, unpin: bool = False
    ):
        """Pin or unpin a message in a chat.

        Args:
            stream_id (int | str): ID of the stream in base 10 or 36.
            message (int): Converting this to int must return a chat message ID.
            unpin (bool): If true, unpins a message instead of pinning it.
        """

        self.sphp_request(
            f"chat.message.{"un" * unpin}pin",
            data={
                "video_id": utils.ensure_b10(stream_id),
                "message_id": int(message),
            },
        )

    def mute_user(
        self,
        username: str,
        is_channel: bool = False,
        video: Optional[SupportsInt] = None,
        duration: Optional[int] = None,
        total: bool = False,
    ):
        """Mute a user or channel by name.

        Args:
            username (str): The user to mute.
            is_channel (bool): Is this a channel name rather than a username?
            video (SupportsInt): The video to mute the user on.
                Defaults to None.
            duration (int): How long the user will be muted for, in seconds.
                Defaults to None, mute permanently.
            total (bool): Is this a mute across all videos?
                Defaults to False, requires video if False.
        """

        self.sphp_request(
            "moderation.mute",
            data={
                "user_to_mute": username,
                "entity_type": ("user", "channel")[is_channel],
                "video": int(video) if video else None,
                "duration": duration,
                "type": ("video", "total")[total],
            },
        )

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

        Args:
            record_id: The numeric ID of the mute record to undo.
        """

        self.sphp_request(
            "moderation.unmute",
            data={
                "record_id": int(record_id),
            },
        )

    def _is_comment_elem(self, e: bs4.Tag) -> bool:
        """Check if a beautifulsoup element is a comment.

        Args:
            e (bs4.Tag): The BeautifulSoup element to test.

        Returns:
            Result (bool): Did the element fit the criteria for being a comment?
        """

        return (
            e.name == "li"
            and "comment-item" in e.get("class")
            and "comments-create" not in e.get("class")
        )

    def comment_list(self, video_id: str | SupportsInt) -> list[scraping.HTMLComment]:
        """Get the list of comments under a video.

        Args:
            video_id (str | SupportsInt): The numeric ID of a video in base 10 or 36.

        Returns:
            Comments (list[scraping.HTMLComment]): A list of scraping.HTMLComment objects.
        """

        r = self.sphp_request(
            "comment.list",
            additional_params={
                "video": utils.ensure_b36(video_id),
            },
            method="GET",
        )
        soup = bs4.BeautifulSoup(r.json()["html"], features="html.parser")
        comment_elems = soup.find_all(self._is_comment_elem)
        return [scraping.HTMLComment(e, self) for e in comment_elems]

    def comment_add(
        self, video_id: SupportsInt | str, comment: str, reply_id: SupportsInt = 0
    ) -> APIComment:
        """Post a comment on a video.

        Args:
            video_id (SupportsInt | str): The numeric ID of a video / stream in base 10 or 36.
            comment (str): What to say.
            reply_id (SupportsInt): The ID of the comment to reply to.
                Defaults to zero (don't reply to anybody).

        Returns:
            Comment (APIComment): The comment, as parsed from the response data.
        """

        r = self.sphp_request(
            "comment.add",
            data={
                "video": utils.ensure_b10(video_id),
                "reply_id": int(reply_id),
                "comment": str(comment),
                "target": "comment-create-1",
            },
        )
        return APIComment(r.json()["data"], self)

    def comment_pin(self, comment_id: SupportsInt, unpin: bool = False):
        """Pin or unpin a comment by ID.

        Args:
            comment_id (SupportsInt): The numeric ID of the comment to pin/unpin.
            unpin (bool): If true, unpins instead of pinning comment.
        """

        self.sphp_request(
            f"comment.{"un" * unpin}pin",
            data={"comment_id": int(comment_id)},
        )

    def comment_delete(self, comment_id: SupportsInt):
        """Delete a comment by ID.

        Args:
            comment_id (SupportsInt): The numeric ID of the comment to delete.
        """

        self.sphp_request(
            "comment.delete",
            data={"comment_id": int(comment_id)},
        )

    def comment_restore(self, comment_id: SupportsInt) -> APIComment:
        """Restore a deleted comment by ID.

        Args:
            comment_id (SupportsInt): The numeric ID of the comment to restore.

        Returns:
            comment (APIComment): The restored comment.
        """

        r = self.sphp_request(
            "comment.restore",
            data={"comment_id": int(comment_id)},
        )
        return APIComment(r.json()["data"], self)

    def get_video_url(self, video_id: SupportsInt | str) -> str:
        """Get the URL of a Rumble video.

        Args:
            video_id (SupportsInt | str): The numeric ID of the video.

        Returns:
            URL (str): The URL of the video.
        """

        r = self.sphp_request(
            "media.share",
            additional_params={
                "video": utils.ensure_b36(video_id),
                "start": 0,
            },
            method="GET",
        )
        soup = bs4.BeautifulSoup(r.json()["html"], features="html.parser")
        elem = soup.find("div", attrs={"class": "fb-share-button share-fb"})
        return elem.attrs["data-url"]

    def playlist_add_video(self, playlist_id: str, video_id: SupportsInt | str):
        """Add a video to a playlist.

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

        self.sphp_request(
            "playlist.add_video",
            data={
                "playlist_id": str(playlist_id),
                "video_id": utils.ensure_b10(video_id),
            },
        )

    def playlist_delete_video(self, playlist_id: str, video_id: SupportsInt | str):
        """Remove a video from a playlist.

        Args:
            playlist_id (str): The numeric ID of the playlist in base 64.
            video_id (int, str): The numeric ID of the video to remove, in base 10 or 36.
        """

        self.sphp_request(
            "playlist.delete_video",
            data={
                "playlist_id": str(playlist_id),
                "video_id": utils.ensure_b10(video_id),
            },
        )

    def playlist_add(
        self,
        title: str,
        description: str = "",
        visibility: str = "public",
        channel_id: SupportsInt | str = None,
    ) -> APIPlaylist:
        """Create a new playlist.

        Args:
            title (str): The title of the playlist.
            description (str): Describe the playlist.
                Defaults to nothing.
            visibility (str): Set to public, unlisted, or private via string.
                Defaults to 'public'.
            channel_id (SupportsInt | str): The numeric ID of the channel to create the playlist under.
                Defaults to none.

        Returns:
            Playlist (APIPlaylist): The playlist as parsed from the response data.
        """

        r = self.sphp_request(
            "playlist.add",
            data={
                "title": str(title),
                "description": str(description),
                "visibility": str(visibility),
                "channel_id": str(utils.ensure_b10(channel_id)) if channel_id else None,
            },
        )
        return APIPlaylist(r.json()["data"], self)

    def playlist_edit(
        self,
        playlist_id: str,
        title: str,
        description: str = "",
        visibility: str = "public",
        channel_id: Optional[SupportsInt | str] = None,
    ) -> APIPlaylist:
        """Edit the details of an existing playlist

        Args:
            playlist_id (str): The numeric ID of the playlist to edit in base 64.
            title (str): The title of the playlist.
            description (str): Describe the playlist.
                Defaults to nothing.
            visibility (str): Set to public, unlisted, or private via string.
                Defaults to 'public'.
            channel_id (SupportsInt | str): The numeric ID of the channel to have the playlist under.
                Defaults to None.

        Returns:
            Playlist (APIPlaylist): The playlist as parsed from the response data.
        """

        r = self.sphp_request(
            "playlist.edit",
            data={
                "title": str(title),
                "description": str(description),
                "visibility": str(visibility),
                "channel_id": str(utils.ensure_b10(channel_id)) if channel_id else None,
                "playlist_id": str(playlist_id),
            },
        )
        return APIPlaylist(r.json()["data"], self)

    def playlist_delete(self, playlist_id: str):
        """Delete a playlist.

        Args:
            playlist_id (str): The numeric ID of the playlist to delete in base 64.
        """

        self.sphp_request(
            "playlist.delete",
            data={"playlist_id": str(playlist_id)},
        )

    def raid_confirm(self, stream_id: SupportsInt | str):
        """Confirm a raid, previously set up by command.

        Args:
            stream_id (SupportsInt | str): The numeric ID of the stream to confirm the raid from, in base 10 or 36.
        """

        self.sphp_request(
            "raid.confirm",
            data={"video_id": utils.ensure_b10(stream_id)},
        )

    def reset_rls_api_key(self, channel_id: SupportsInt | str = None) -> bool:
        """
        Reset the Live Stream API URL for a user or channel.

        Args:
            channel_id (SupportsInt | str): The channel to reset the API key for.
                Defaults to None, reset the user API key.

        Returns:
            success (bool): The success attribute of the returned JSON.
        """

        return self.sphp_request(
            "livestream_api.reset_api_key",
            data={
                "channel_id": str(utils.ensure_b10(channel_id)) if channel_id else None
            },
        ).json().get("data", {}).get("success", False)

session_cookie = None instance-attribute

The session token cookie we received on login. Can be reused if not logged out.

session_token property

The token stored in our session cookie, empty if if we are not logged in

user_id property

The numeric ID of the logged in user in base 10

user_id_b10 property

The numeric ID of the logged in user in base 10

user_id_b36 property

The numeric ID of the logged in user in base 36

username = username instance-attribute

The username this ServicePHP wrapper is working under

__init__(username=None, session=None)

Interact with Rumble's service.php API.

Parameters:

Name	Type	Description	Default
username	str	The username we will be working under. Defaults to None, refer to session.	None
session	str | dict	A pre-existing session token. Defaults to None, await login.	None

Source code in cocorum/servicephp.py

def __init__(self, username: str = None, session: str | dict = None):
    """Interact with Rumble's service.php API.

    Args:
        username (str): The username we will be working under.
            Defaults to None, refer to session.
        session (str | dict): A pre-existing session token.
            Defaults to None, await login.
    """

    assert username or session, "Must pass at least one thing to log in with"

    # Set up initial auth data variables
    self.username: str | None = username
    """The username this ServicePHP wrapper is working under"""

    self.session_cookie: dict | None = None
    """The session token cookie we received on login. Can be reused if not logged out."""

    # Session is the token directly
    if isinstance(session, str):
        self.session_cookie = {static.Misc.session_token_key: session}

    # Session is a cookie dict
    elif isinstance(session, dict):
        assert session.get(
            static.Misc.session_token_key
        ), f"Session cookie dict must have '{static.Misc.session_token_key}' as key."
        self.session_cookie = session

    # Session was passed but it is not anything we can use
    elif session is not None:
        raise ValueError(
            f"Session must be a token str or cookie dict, got {type(session)}"
        )

    # If a session cookie was passed, test it
    assert not self.session_cookie or utils.test_session_cookie(
        self.session_cookie
    ), "Session cookie is invalid."

    # Stored ID of the logged in user
    self.__user_id = None

__repr__()

String to represent this object

Source code in cocorum/servicephp.py

def __repr__(self) -> str:
    """String to represent this object"""
    return f"{type(self).__name__}(username={self.username}, <{'not ' * bool(self.session_cookie)}logged in>)"

chat_pin(stream_id, message, unpin=False)

Pin or unpin a message in a chat.

Parameters:

Name	Type	Description	Default
stream_id	int | str	ID of the stream in base 10 or 36.	required
message	int	Converting this to int must return a chat message ID.	required
unpin	bool	If true, unpins a message instead of pinning it.	False

Source code in cocorum/servicephp.py

def chat_pin(
    self, stream_id: SupportsInt | str, message: SupportsInt, unpin: bool = False
):
    """Pin or unpin a message in a chat.

    Args:
        stream_id (int | str): ID of the stream in base 10 or 36.
        message (int): Converting this to int must return a chat message ID.
        unpin (bool): If true, unpins a message instead of pinning it.
    """

    self.sphp_request(
        f"chat.message.{"un" * unpin}pin",
        data={
            "video_id": utils.ensure_b10(stream_id),
            "message_id": int(message),
        },
    )

comment_add(video_id, comment, reply_id=0)

Post a comment on a video.

Parameters:

Name	Type	Description	Default
video_id	SupportsInt | str	The numeric ID of a video / stream in base 10 or 36.	required
comment	str	What to say.	required
reply_id	SupportsInt	The ID of the comment to reply to. Defaults to zero (don't reply to anybody).	0

Returns:

Name	Type	Description
Comment	APIComment	The comment, as parsed from the response data.

Source code in cocorum/servicephp.py

def comment_add(
    self, video_id: SupportsInt | str, comment: str, reply_id: SupportsInt = 0
) -> APIComment:
    """Post a comment on a video.

    Args:
        video_id (SupportsInt | str): The numeric ID of a video / stream in base 10 or 36.
        comment (str): What to say.
        reply_id (SupportsInt): The ID of the comment to reply to.
            Defaults to zero (don't reply to anybody).

    Returns:
        Comment (APIComment): The comment, as parsed from the response data.
    """

    r = self.sphp_request(
        "comment.add",
        data={
            "video": utils.ensure_b10(video_id),
            "reply_id": int(reply_id),
            "comment": str(comment),
            "target": "comment-create-1",
        },
    )
    return APIComment(r.json()["data"], self)

comment_delete(comment_id)

Delete a comment by ID.

Parameters:

Name	Type	Description	Default
comment_id	SupportsInt	The numeric ID of the comment to delete.	required

Source code in cocorum/servicephp.py

def comment_delete(self, comment_id: SupportsInt):
    """Delete a comment by ID.

    Args:
        comment_id (SupportsInt): The numeric ID of the comment to delete.
    """

    self.sphp_request(
        "comment.delete",
        data={"comment_id": int(comment_id)},
    )

comment_list(video_id)

Get the list of comments under a video.

Parameters:

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

Returns:

Name	Type	Description
Comments	list[HTMLComment]	A list of scraping.HTMLComment objects.

Source code in cocorum/servicephp.py

def comment_list(self, video_id: str | SupportsInt) -> list[scraping.HTMLComment]:
    """Get the list of comments under a video.

    Args:
        video_id (str | SupportsInt): The numeric ID of a video in base 10 or 36.

    Returns:
        Comments (list[scraping.HTMLComment]): A list of scraping.HTMLComment objects.
    """

    r = self.sphp_request(
        "comment.list",
        additional_params={
            "video": utils.ensure_b36(video_id),
        },
        method="GET",
    )
    soup = bs4.BeautifulSoup(r.json()["html"], features="html.parser")
    comment_elems = soup.find_all(self._is_comment_elem)
    return [scraping.HTMLComment(e, self) for e in comment_elems]

comment_pin(comment_id, unpin=False)

Pin or unpin a comment by ID.

Parameters:

Name	Type	Description	Default
comment_id	SupportsInt	The numeric ID of the comment to pin/unpin.	required
unpin	bool	If true, unpins instead of pinning comment.	False

Source code in cocorum/servicephp.py

def comment_pin(self, comment_id: SupportsInt, unpin: bool = False):
    """Pin or unpin a comment by ID.

    Args:
        comment_id (SupportsInt): The numeric ID of the comment to pin/unpin.
        unpin (bool): If true, unpins instead of pinning comment.
    """

    self.sphp_request(
        f"comment.{"un" * unpin}pin",
        data={"comment_id": int(comment_id)},
    )

comment_restore(comment_id)

Restore a deleted comment by ID.

Parameters:

Name	Type	Description	Default
comment_id	SupportsInt	The numeric ID of the comment to restore.	required

Returns:

Name	Type	Description
comment	APIComment	The restored comment.

Source code in cocorum/servicephp.py

def comment_restore(self, comment_id: SupportsInt) -> APIComment:
    """Restore a deleted comment by ID.

    Args:
        comment_id (SupportsInt): The numeric ID of the comment to restore.

    Returns:
        comment (APIComment): The restored comment.
    """

    r = self.sphp_request(
        "comment.restore",
        data={"comment_id": int(comment_id)},
    )
    return APIComment(r.json()["data"], self)

get_2fa_first_step(password_hashes)

Get 2FA info on our account as a first 2FA step

Parameters:

Name	Type	Description	Default
password_hashes	str	The hashed password.	required

Returns:

Name	Type	Description
info	TwoFacAuth	Information on 2FA.

Source code in cocorum/servicephp.py

def get_2fa_first_step(self, password_hashes: str) -> TwoFacAuth:
    """Get 2FA info on our account as a first 2FA step

    Args:
        password_hashes (str): The hashed password.

    Returns:
        info (TwoFacAuth): Information on 2FA."""

    r = self.sphp_request(
        "user.2fa.first_step",
        data={
            "legacy_password": 1,  # TODO
            "login": self.username,
            "password": password_hashes,
            "redirect_url": static.URI.rumble_base,
        },
        logged_in=False,
    )

    return TwoFacAuth(r.json()["data"], self)

get_hashed_password(password)

Use Rumble's salts to hash a pasword for login

Parameters:

Name	Type	Description	Default
password	str	The password to hash with the salts.	required

Returns:

Name	Type	Description
hashed	str	The hashed password to use in the rest of the login.

Source code in cocorum/servicephp.py

def get_hashed_password(self, password: str) -> str:
    """Use Rumble's salts to hash a pasword for login

    Args:
        password (str): The password to hash with the salts.

    Returns:
        hashed (str): The hashed password to use in the rest of the login.
    """

    # Get salts
    r = self.sphp_request(
        "user.get_salts",
        data={"username": self.username},
        logged_in=False,
        additional_params={"response_type": "session"},
    )
    salts = r.json()["data"]["salts"]

    return ",".join(utils.calc_password_hashes(password, salts))

get_video_url(video_id)

Get the URL of a Rumble video.

Parameters:

Name	Type	Description	Default
video_id	SupportsInt | str	The numeric ID of the video.	required

Returns:

Name	Type	Description
URL	str	The URL of the video.

Source code in cocorum/servicephp.py

def get_video_url(self, video_id: SupportsInt | str) -> str:
    """Get the URL of a Rumble video.

    Args:
        video_id (SupportsInt | str): The numeric ID of the video.

    Returns:
        URL (str): The URL of the video.
    """

    r = self.sphp_request(
        "media.share",
        additional_params={
            "video": utils.ensure_b36(video_id),
            "start": 0,
        },
        method="GET",
    )
    soup = bs4.BeautifulSoup(r.json()["html"], features="html.parser")
    elem = soup.find("div", attrs={"class": "fb-share-button share-fb"})
    return elem.attrs["data-url"]

login_basic(password)

Perform a basic password authentication as the first step of login. If successful, sets our session token and returns None. If incomplete, returns TwoFacAuth() for next step.

Parameters:

Name	Type	Description	Default
password	str	The password to login with.	required

Returns:

Name	Type	Description
result	TwoFacAuth | None	The result of the first login step. None means login is complete.

Source code in cocorum/servicephp.py

def login_basic(self, password: str) -> TwoFacAuth | None:
    """Perform a basic password authentication as the first step of login.
        If successful, sets our session token and returns None.
        If incomplete, returns TwoFacAuth() for next step.

    Args:
        password (str): The password to login with.

    Returns:
        result (TwoFacAuth | None): The result of the first login step.
            None means login is complete.
    """

    # Get the password hash to start off
    ph = self.get_hashed_password(password)

    # Check for 2FA
    two_fac_info = self.get_2fa_first_step(ph)

    # 2FA is enabled
    if two_fac_info.totp_id:  # two_fac_auth.options:
        return two_fac_info

    # Get session token for non-2FA
    r = self.sphp_request(
        "user.login",
        data={
            "username": self.username,
            # Hash the password using the salts
            "password_hashes": ph,
        },
        logged_in=False,
    )

    j = r.json()
    session_token = j["data"]["session"]
    assert session_token, f"Login failed: No token returned\n{r.json()}"

    self.session_cookie = {static.Misc.session_token_key: session_token}
    return None

login_second_factor(two_fac_auth, code)

Complete login with a 2FA code.

Parameters:

Name	Type	Description	Default
two_fac_auth	TwoFacAuth	2FA information on this account.	required
code	str	The code to enter.	required

Source code in cocorum/servicephp.py

def login_second_factor(self, two_fac_auth: TwoFacAuth, code: str):
    """Complete login with a 2FA code.

    Args:
        two_fac_auth (TwoFacAuth): 2FA information on this account.
        code (str): The code to enter."""

    # If this fails, sphp_request will raise AssertionError
    r = self.sphp_request(
        "user.2fa.verify_totp",
        data={
            "code": code,
            "redirect_url": static.URI.rumble_base,
            "totp_id": two_fac_auth.totp_id,
            "user_key": two_fac_auth.user_key,
        },
        logged_in=False,
    )

    cookie_start = static.Misc.session_token_key + "="
    for piece in r.headers["Set-Cookie"].split():
        if piece.startswith(cookie_start):
            self.session_cookie = {
                static.Misc.session_token_key: piece.removeprefix(
                    cookie_start
                ).removesuffix(";")
            }
            return
    raise ValueError(
        "Login failed: Did not find session token in 2FA response headers."
    )

logout()

Log out this ServicePHP instance

Source code in cocorum/servicephp.py

def logout(self):
    """Log out this ServicePHP instance"""
    self.sphp_request("user.logout", method="GET")
    self.session_cookie = None

mute_user(username, is_channel=False, video=None, duration=None, total=False)

Mute a user or channel by name.

Parameters:

Name	Type	Description	Default
username	str	The user to mute.	required
is_channel	bool	Is this a channel name rather than a username?	False
video	SupportsInt	The video to mute the user on. Defaults to None.	None
duration	int	How long the user will be muted for, in seconds. Defaults to None, mute permanently.	None
total	bool	Is this a mute across all videos? Defaults to False, requires video if False.	False

Source code in cocorum/servicephp.py

def mute_user(
    self,
    username: str,
    is_channel: bool = False,
    video: Optional[SupportsInt] = None,
    duration: Optional[int] = None,
    total: bool = False,
):
    """Mute a user or channel by name.

    Args:
        username (str): The user to mute.
        is_channel (bool): Is this a channel name rather than a username?
        video (SupportsInt): The video to mute the user on.
            Defaults to None.
        duration (int): How long the user will be muted for, in seconds.
            Defaults to None, mute permanently.
        total (bool): Is this a mute across all videos?
            Defaults to False, requires video if False.
    """

    self.sphp_request(
        "moderation.mute",
        data={
            "user_to_mute": username,
            "entity_type": ("user", "channel")[is_channel],
            "video": int(video) if video else None,
            "duration": duration,
            "type": ("video", "total")[total],
        },
    )

playlist_add(title, description='', visibility='public', channel_id=None)

Create a new playlist.

Parameters:

Name	Type	Description	Default
title	str	The title of the playlist.	required
description	str	Describe the playlist. Defaults to nothing.	''
visibility	str	Set to public, unlisted, or private via string. Defaults to 'public'.	'public'
channel_id	SupportsInt | str	The numeric ID of the channel to create the playlist under. Defaults to none.	None

Returns:

Name	Type	Description
Playlist	APIPlaylist	The playlist as parsed from the response data.

Source code in cocorum/servicephp.py

def playlist_add(
    self,
    title: str,
    description: str = "",
    visibility: str = "public",
    channel_id: SupportsInt | str = None,
) -> APIPlaylist:
    """Create a new playlist.

    Args:
        title (str): The title of the playlist.
        description (str): Describe the playlist.
            Defaults to nothing.
        visibility (str): Set to public, unlisted, or private via string.
            Defaults to 'public'.
        channel_id (SupportsInt | str): The numeric ID of the channel to create the playlist under.
            Defaults to none.

    Returns:
        Playlist (APIPlaylist): The playlist as parsed from the response data.
    """

    r = self.sphp_request(
        "playlist.add",
        data={
            "title": str(title),
            "description": str(description),
            "visibility": str(visibility),
            "channel_id": str(utils.ensure_b10(channel_id)) if channel_id else None,
        },
    )
    return APIPlaylist(r.json()["data"], self)

playlist_add_video(playlist_id, video_id)

Add a video to a playlist.

Parameters:

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

Source code in cocorum/servicephp.py

def playlist_add_video(self, playlist_id: str, video_id: SupportsInt | str):
    """Add a video to a playlist.

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

    self.sphp_request(
        "playlist.add_video",
        data={
            "playlist_id": str(playlist_id),
            "video_id": utils.ensure_b10(video_id),
        },
    )

playlist_delete(playlist_id)

Delete a playlist.

Parameters:

Name	Type	Description	Default
playlist_id	str	The numeric ID of the playlist to delete in base 64.	required

Source code in cocorum/servicephp.py

def playlist_delete(self, playlist_id: str):
    """Delete a playlist.

    Args:
        playlist_id (str): The numeric ID of the playlist to delete in base 64.
    """

    self.sphp_request(
        "playlist.delete",
        data={"playlist_id": str(playlist_id)},
    )

playlist_delete_video(playlist_id, video_id)

Remove a video from a playlist.

Parameters:

Name	Type	Description	Default
playlist_id	str	The numeric ID of the playlist in base 64.	required
video_id	(int, str)	The numeric ID of the video to remove, in base 10 or 36.	required

Source code in cocorum/servicephp.py

def playlist_delete_video(self, playlist_id: str, video_id: SupportsInt | str):
    """Remove a video from a playlist.

    Args:
        playlist_id (str): The numeric ID of the playlist in base 64.
        video_id (int, str): The numeric ID of the video to remove, in base 10 or 36.
    """

    self.sphp_request(
        "playlist.delete_video",
        data={
            "playlist_id": str(playlist_id),
            "video_id": utils.ensure_b10(video_id),
        },
    )

playlist_edit(playlist_id, title, description='', visibility='public', channel_id=None)

Edit the details of an existing playlist

Parameters:

Name	Type	Description	Default
playlist_id	str	The numeric ID of the playlist to edit in base 64.	required
title	str	The title of the playlist.	required
description	str	Describe the playlist. Defaults to nothing.	''
visibility	str	Set to public, unlisted, or private via string. Defaults to 'public'.	'public'
channel_id	SupportsInt | str	The numeric ID of the channel to have the playlist under. Defaults to None.	None

Returns:

Name	Type	Description
Playlist	APIPlaylist	The playlist as parsed from the response data.

Source code in cocorum/servicephp.py

def playlist_edit(
    self,
    playlist_id: str,
    title: str,
    description: str = "",
    visibility: str = "public",
    channel_id: Optional[SupportsInt | str] = None,
) -> APIPlaylist:
    """Edit the details of an existing playlist

    Args:
        playlist_id (str): The numeric ID of the playlist to edit in base 64.
        title (str): The title of the playlist.
        description (str): Describe the playlist.
            Defaults to nothing.
        visibility (str): Set to public, unlisted, or private via string.
            Defaults to 'public'.
        channel_id (SupportsInt | str): The numeric ID of the channel to have the playlist under.
            Defaults to None.

    Returns:
        Playlist (APIPlaylist): The playlist as parsed from the response data.
    """

    r = self.sphp_request(
        "playlist.edit",
        data={
            "title": str(title),
            "description": str(description),
            "visibility": str(visibility),
            "channel_id": str(utils.ensure_b10(channel_id)) if channel_id else None,
            "playlist_id": str(playlist_id),
        },
    )
    return APIPlaylist(r.json()["data"], self)

raid_confirm(stream_id)

Confirm a raid, previously set up by command.

Parameters:

Name	Type	Description	Default
stream_id	SupportsInt | str	The numeric ID of the stream to confirm the raid from, in base 10 or 36.	required

Source code in cocorum/servicephp.py

def raid_confirm(self, stream_id: SupportsInt | str):
    """Confirm a raid, previously set up by command.

    Args:
        stream_id (SupportsInt | str): The numeric ID of the stream to confirm the raid from, in base 10 or 36.
    """

    self.sphp_request(
        "raid.confirm",
        data={"video_id": utils.ensure_b10(stream_id)},
    )

reset_rls_api_key(channel_id=None)

Reset the Live Stream API URL for a user or channel.

Parameters:

Name	Type	Description	Default
channel_id	SupportsInt | str	The channel to reset the API key for. Defaults to None, reset the user API key.	None

Returns:

Name	Type	Description
success	bool	The success attribute of the returned JSON.

Source code in cocorum/servicephp.py

def reset_rls_api_key(self, channel_id: SupportsInt | str = None) -> bool:
    """
    Reset the Live Stream API URL for a user or channel.

    Args:
        channel_id (SupportsInt | str): The channel to reset the API key for.
            Defaults to None, reset the user API key.

    Returns:
        success (bool): The success attribute of the returned JSON.
    """

    return self.sphp_request(
        "livestream_api.reset_api_key",
        data={
            "channel_id": str(utils.ensure_b10(channel_id)) if channel_id else None
        },
    ).json().get("data", {}).get("success", False)

sphp_request(service_name, data={}, additional_params={}, logged_in=True, method='POST')

Make a request to Service.PHP with common settings

Parameters:

Name	Type	Description	Default
service_name	str	The name parameter of the specific PHP service	required
data	dict	Form data. Defaults to {}.	{}
additional_params	dict	Any additional query string parameters. Defaults to {}.	{}
logged_in	bool	Should the request use the session cookie? Defaults to True.	True
method	str	What request method to use. Defaults to 'POST'.	'POST'

Returns:

Name	Type	Description
response	Response	The response from Rumble.

Source code in cocorum/servicephp.py

def sphp_request(
    self,
    service_name: str,
    data: dict = {},
    additional_params: dict = {},
    logged_in: bool = True,
    method: str = "POST",
) -> requests.Response:
    """
    Make a request to Service.PHP with common settings

    Args:
        service_name (str): The name parameter of the specific PHP service
        data (dict): Form data.
            Defaults to {}.
        additional_params (dict): Any additional query string parameters.
            Defaults to {}.
        logged_in (bool): Should the request use the session cookie?
            Defaults to True.
        method (str): What request method to use.
            Defaults to 'POST'.

    Returns:
        response (requests.Response): The response from Rumble.
    """
    assert self.session_cookie or not logged_in, "Not logged in yet."

    params = {"name": service_name}
    params.update(additional_params)
    r = requests.request(
        method,
        static.URI.servicephp,
        params=params,
        data=data,
        headers=static.RequestHeaders.user_agent,
        cookies=self.session_cookie if logged_in else None,
        timeout=static.Delays.request_timeout,
    )
    assert (
        r.status_code == 200
    ), f"Service.PHP request for {service_name} failed: {r}\n{r.text}"
    # If the request json has a data -> success value, make sure it is True
    d = r.json().get("data")
    if isinstance(d, dict):
        assert d.get(
            "success", True
        ), f"Service.PHP request for {service_name} failed: \n{r.text}"
    # Data was not a dict but was not empty
    elif d:
        print(
            f"Service.PHP request for {service_name} did not fail but returned unknown data type {type(d)}: {d}"
        )

    return r

unmute_user(record_id)

Unmute a user.

Parameters:

Name	Type	Description	Default
record_id	SupportsInt	The numeric ID of the mute record to undo.	required

Source code in cocorum/servicephp.py

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

    Args:
        record_id: The numeric ID of the mute record to undo.
    """

    self.sphp_request(
        "moderation.unmute",
        data={
            "record_id": int(record_id),
        },
    )

TwoFacAuth

Bases: JSONObj

Handle and use the resposne data from the 2FA request

Source code in cocorum/servicephp.py

class TwoFacAuth(JSONObj):
    """Handle and use the resposne data from the 2FA request"""

    def __init__(self, jsondata: dict, servicephp: ServicePHP):
        """Handle and use the resposne data from the 2FA request.

        Args:
            jsondata (dict): The JSON data block of the 2FA first step data.
            servicephp (ServicePHP): The ServicePHP object that spawned us.
        """

        JSONObj.__init__(self, jsondata)
        self.servicephp: ServicePHP = servicephp
        """Our parent ServicePHP wrapper"""

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

    @property
    def totp_id(self) -> str | None:
        """Is None when 2FA is disabled"""
        return self["totp_id"]

    @property
    def user_key(self) -> str | None:
        """Is None when 2FA is disabled"""
        return self.get("user_key")

    @property
    def options(self) -> tuple:
        """The options we have for 2FA"""
        # 2FA is disabled it totp_id is None, so we have no options
        if not self.totp_id:
            return tuple()

        return tuple(
            option for option, available in self["options"].items() if available
        )

    def request_2fa_code(self, option: str = "authenticator") -> str | None:
        """Tell Rumble to send the user a 2FA code

        Args:
            option (str): The 2FA option to use. Must be in self.options.
                Defaults to "authenticator" app.

        Returns:
            sent_to (str | None): The code destination, semi-censored.
                If 'authenticator' was used, returns None, as such apps
                generate the 2FA codes on demand."""

        assert self.options, "2FA is not enabled on this account"

        assert (
            option in self.options
        ), f"Option '{option}' is not enabled on this account for 2FA"

        if option == "email":
            r = self.servicephp.sphp_request(
                "user.2fa.request_email_code",
                data={
                    "totp_id": self.totp_id,
                    "user_key": self.user_key,
                },
                logged_in=False,
            )
            return r.json()["data"]["sent_to"]

        if option == "phone":
            r = self.servicephp.sphp_request(
                "user.2fa.request_sms_code",
                data={
                    "totp_id": self.totp_id,
                    "user_key": self.user_key,
                },
                logged_in=False,
            )
            return r.json()["data"]["sent_to"]

        if option == "authenticator":
            # Nothing for us to do
            return None

        raise NotImplementedError(
            f"Cocorum does not support option '{option}' for 2FA."
        )

options property

The options we have for 2FA

servicephp = servicephp instance-attribute

Our parent ServicePHP wrapper

totp_id property

Is None when 2FA is disabled

user_key property

Is None when 2FA is disabled

__init__(jsondata, servicephp)

Handle and use the resposne data from the 2FA request.

Parameters:

Name	Type	Description	Default
jsondata	dict	The JSON data block of the 2FA first step data.	required
servicephp	ServicePHP	The ServicePHP object that spawned us.	required

Source code in cocorum/servicephp.py

def __init__(self, jsondata: dict, servicephp: ServicePHP):
    """Handle and use the resposne data from the 2FA request.

    Args:
        jsondata (dict): The JSON data block of the 2FA first step data.
        servicephp (ServicePHP): The ServicePHP object that spawned us.
    """

    JSONObj.__init__(self, jsondata)
    self.servicephp: ServicePHP = servicephp
    """Our parent ServicePHP wrapper"""

__repr__()

String to represent this object

Source code in cocorum/servicephp.py

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

request_2fa_code(option='authenticator')

Tell Rumble to send the user a 2FA code

Parameters:

Name	Type	Description	Default
option	str	The 2FA option to use. Must be in self.options. Defaults to "authenticator" app.	'authenticator'

Returns:

Name	Type	Description
sent_to	str | None	The code destination, semi-censored. If 'authenticator' was used, returns None, as such apps generate the 2FA codes on demand.

Source code in cocorum/servicephp.py

def request_2fa_code(self, option: str = "authenticator") -> str | None:
    """Tell Rumble to send the user a 2FA code

    Args:
        option (str): The 2FA option to use. Must be in self.options.
            Defaults to "authenticator" app.

    Returns:
        sent_to (str | None): The code destination, semi-censored.
            If 'authenticator' was used, returns None, as such apps
            generate the 2FA codes on demand."""

    assert self.options, "2FA is not enabled on this account"

    assert (
        option in self.options
    ), f"Option '{option}' is not enabled on this account for 2FA"

    if option == "email":
        r = self.servicephp.sphp_request(
            "user.2fa.request_email_code",
            data={
                "totp_id": self.totp_id,
                "user_key": self.user_key,
            },
            logged_in=False,
        )
        return r.json()["data"]["sent_to"]

    if option == "phone":
        r = self.servicephp.sphp_request(
            "user.2fa.request_sms_code",
            data={
                "totp_id": self.totp_id,
                "user_key": self.user_key,
            },
            logged_in=False,
        )
        return r.json()["data"]["sent_to"]

    if option == "authenticator":
        # Nothing for us to do
        return None

    raise NotImplementedError(
        f"Cocorum does not support option '{option}' for 2FA."
    )

S.D.G.