TikTokLive.client.ws package

Submodules

TikTokLive.client.ws.ws_client module

class TikTokLive.client.ws.ws_client.WebcastWSClient(ws_kwargs: dict | None = None, ws_proxy: Proxy | AsyncioProxy | None = None)

Bases: object

Websocket client responsible for connections to TikTok

DEFAULT_PING_INTERVAL: float = 1.0
PING_MESSAGE: bytes = b'2\x02pb:\x02hb'
async connect(room_id: int, cookies: Cookies, authenticate_websocket: bool, user_agent: str, initial_webcast_response: ProtoMessageFetchResult, process_connect_events: bool = True, compress_ws_events: bool = True) → AsyncIterator[ProtoMessageFetchResult]

Connect to the Webcast server & iterate over response messages.

— Message 1 —

The iterator exits normally when the connection is closed with close code 1000 (OK) or 1001 (going away) or without a close code. It raises a ConnectionClosedError when the connection is closed with any other code.

— Message 2 —

DEVELOP SANITY NOTE:

When ping_timeout is set (i.e. not None), the client waits for a pong for N seconds. TikTok DO NOT SEND pongs back. Unfortunately the websockets client after N seconds assumes the server is dead. It then throws the following infamous exception:

websockets.exceptions.ConnectionClosedError: sent 1011 (unexpected error) keepalive ping timeout; no close frame received

If you set ping_timeout to None, it doesn’t wait for a pong. Perfect, since TikTok don’t send them.

— Parameters –

Parameters:

  • initial_webcast_response – The Initial ProtoMessageFetchResult from the sign server - NOT a PushFrame

  • room_id – The room ID to connect to

  • user_agent – The user agent to pass to the WebSocket connection

  • cookies – The cookies to pass to the WebSocket connection

  • authenticate_websocket – Whether to authenticate the WebSocket connection

  • process_connect_events – Whether to process the initial events sent in the first fetch

  • compress_ws_events – Whether to ask TikTok to gzip the WebSocket events

Returns:

Yields ProtoMessageFetchResultMessage, the messages within ProtoMessageFetchResult.messages

property connected: bool

Check if the WebSocket is open

Returns:

WebSocket status

async disconnect() → None

Request to stop the websocket connection & wait :return: None

Get the cookie string for the WebSocket connection.

Parameters:

  • cookies – Cookies to pass to the WebSocket connection

  • authenticate_websocket – Whether the WebSocket is authenticated

Returns:

Cookie string

restart_ping_loop() → None

Restart the WebSocket ping loop

async send(message: bytes | Message) → None

Send a message to the WebSocket

Parameters:

message – Message to send to the WebSocket connection

async send_ack(webcast_response: ProtoMessageFetchResult, webcast_push_frame: WebcastPushFrame) → None

Acknowledge the receipt of a ProtoMessageFetchResult message from TikTok, if necessary

Parameters:

  • webcast_response – The ProtoMessageFetchResult to acknowledge

  • webcast_push_frame – The WebcastPushFrame containing the ProtoMessageFetchResult

Returns:

None

property ws: WebSocketClientProtocol | None

Get the current WebSocketClientProtocol

Returns:

WebSocketClientProtocol

TikTokLive.client.ws.ws_connect module

class TikTokLive.client.ws.ws_connect.WebcastConnect(initial_webcast_response: ProtoMessageFetchResult, logger: Logger, base_uri_params: Dict[str, Any], base_uri_append_str: str, uri: str | None = None, **kwargs)

Bases: Connect

property ws: WebSocketClientProtocol | None

Get the current WebSocketClientProtocol

TikTokLive.client.ws.ws_connect.WebcastProxy

Type hint for a WebcastIterator, which yields a tuple of WebcastPushFrame and ProtoMessageFetchResult. WebcastPushFrame is Optional because the first yielded item is from the initial response which is from /im/fetch (from the sign server), so it is not encapsulated by a WebcastPushFrame.

alias of Proxy | AsyncioProxy

class TikTokLive.client.ws.ws_connect.WebcastProxyConnect(proxy: Proxy | AsyncioProxy | None, **kwargs)

Bases: ProxyConnect, WebcastConnect

Add Proxy support to the WebcastConnect class

TikTokLive.client.ws.ws_utils module

TikTokLive.client.ws.ws_utils.build_webcast_uri(initial_webcast_response: ProtoMessageFetchResult, base_uri_params: dict, base_uri_append_str: str) → str

Build a webcast URI from a base URI and parameters. This method will format the base URI with the parameters and return the formatted URI.

Parameters:

  • initial_webcast_response – The initial Webcast response

  • base_uri_params – Parameters to format the URI with

  • base_uri_append_str – String to append to the base URI

Returns:

str The formatted URI

TikTokLive.client.ws.ws_utils.extract_webcast_push_frame(data: bytes, logger: ~logging.Logger = <Logger TikTokLive (WARNING)>) → WebcastPushFrame

Extract a WebcastPushFrame from a raw byte payload. This method will parse the payload and return a WebcastPushFrame object. This method is useful for extracting push frames from a WebSocket connection.

Parameters:

  • data – Raw byte payload to extract from

  • logger – Logger to use for logging

Returns:

WebcastPushFrame The extracted push frame

TikTokLive.client.ws.ws_utils.extract_webcast_response_message(push_frame: ~TikTokLive.proto.custom_extras.WebcastPushFrame, logger: ~logging.Logger = <Logger TikTokLive (WARNING)>) → ProtoMessageFetchResult

Extract the ProtoMessageFetchResult from a push frame. If compression is enabled on the WebSocket, then messages will come gzipped. This method will decompress the payload if necessary. The gzip format allows for less bandwidth usage, at the cost of a slight CPU increase for message decompression.

Parameters:

  • push_frame – Push frame to extract from

  • logger – Logger to use for logging

Returns:

ProtoMessageFetchResult The extracted response

Module contents