TikTokLive.client package

Subpackages

• TikTokLive.client.web package
  • Subpackages
    • TikTokLive.client.web.routes package
      • Submodules
      • TikTokLive.client.web.routes.fetch_gift_list module
      • TikTokLive.client.web.routes.fetch_image module
      • TikTokLive.client.web.routes.fetch_room_id module
      • TikTokLive.client.web.routes.fetch_room_info module
      • TikTokLive.client.web.routes.fetch_sign module
      • TikTokLive.client.web.routes.fetch_video module
      • Module contents
  • Submodules
  • TikTokLive.client.web.web_base module
    • ClientRoute
    • TikTokHTTPClient
      • TikTokHTTPClient.close()
      • TikTokHTTPClient.generate_device_id()
      • TikTokHTTPClient.get_response()
      • TikTokHTTPClient.set_session_id()
  • TikTokLive.client.web.web_client module
    • TikTokWebClient
  • TikTokLive.client.web.web_settings module
  • Module contents
• TikTokLive.client.ws package
  • Submodules
  • TikTokLive.client.ws.ws_client module
    • WebcastProxyConnect
    • WebcastWSClient
      • WebcastWSClient.build_connection_args()
      • WebcastWSClient.connect()
      • WebcastWSClient.connect_loop()
      • WebcastWSClient.connected
      • WebcastWSClient.disconnect()
      • WebcastWSClient.process_recv()
      • WebcastWSClient.send_ack()
  • Module contents

Submodules

TikTokLive.client.client module

class TikTokLive.client.client.TikTokLiveClient(unique_id: str, web_proxy: Proxy | None = None, ws_proxy: Proxy | None = None, web_kwargs: dict = {}, ws_kwargs: dict = {})

Bases: AsyncIOEventEmitter

A client to connect to & read from TikTok LIVE streams

add_listener(event: Type[WebsocketResponseEvent | UnknownEvent | ConnectEvent | FollowEvent | ShareEvent | LiveEndEvent | LivePauseEvent | LiveUnpauseEvent | DisconnectEvent | GiftEvent | RoomEvent | BarrageEvent | CaptionEvent | CommentEvent | ControlEvent | EmoteChatEvent | EnvelopeEvent | GoalUpdateEvent | ImDeleteEvent | LikeEvent | RoomUserSeqEvent | SocialEvent | SubscribeEvent | RankUpdateEvent | JoinEvent | PollEvent | QuestionNewEvent | RankTextEvent | HourlyRankEvent | LinkMicArmiesEvent | LinkMicBattleEvent | LinkMicFanTicketMethodEvent | LinkMicMethodEvent | LiveIntroEvent | UnauthorizedMemberEvent | MessageDetectEvent | OecLiveShoppingEvent | RoomPinEvent | SystemEvent | LinkEvent | LinkLayerEvent], f: EventHandler) → Handler

Method that can be used to register a Python function as an event listener

Parameters:

• event – The event to listen to

• f – The function to handle the event

Returns:

The generated pyee.Handler object

async connect(**kwargs) → Task

Start a future-blocking connection to TikTokLive

Parameters:

kwargs – Kwargs to pass to start

Returns:

The task, once it’s finished

property connected: bool

Whether the WebSocket client is currently connected to TikTok

Returns:

Connection status

async disconnect() → None

Disconnect the client from the websocket

Returns:

None

property gift_info: dict | None

Information about the stream’s gifts if fetch_gift_info=True when starting the client e.g. with client.run)

Returns:

The stream gift info

has_listener(event: Type[WebsocketResponseEvent | UnknownEvent | ConnectEvent | FollowEvent | ShareEvent | LiveEndEvent | LivePauseEvent | LiveUnpauseEvent | DisconnectEvent | GiftEvent | RoomEvent | BarrageEvent | CaptionEvent | CommentEvent | ControlEvent | EmoteChatEvent | EnvelopeEvent | GoalUpdateEvent | ImDeleteEvent | LikeEvent | RoomUserSeqEvent | SocialEvent | SubscribeEvent | RankUpdateEvent | JoinEvent | PollEvent | QuestionNewEvent | RankTextEvent | HourlyRankEvent | LinkMicArmiesEvent | LinkMicBattleEvent | LinkMicFanTicketMethodEvent | LinkMicMethodEvent | LiveIntroEvent | UnauthorizedMemberEvent | MessageDetectEvent | OecLiveShoppingEvent | RoomPinEvent | SystemEvent | LinkEvent | LinkLayerEvent]) → bool

Check whether the client is listening to a given event

Parameters:

event – The event to check listening for

Returns:

Whether it is being listened to

async is_live(unique_id: str | None = None) → bool

Check if the client is currently live on TikTok

Parameters:

unique_id – Optionally override the user to check

Returns:

Whether they are live on TikTok

property logger: Logger

The internal logger used by TikTokLive

Returns:

An instance of a logging.Logger

on(event: Type[WebsocketResponseEvent | UnknownEvent | ConnectEvent | FollowEvent | ShareEvent | LiveEndEvent | LivePauseEvent | LiveUnpauseEvent | DisconnectEvent | GiftEvent | RoomEvent | BarrageEvent | CaptionEvent | CommentEvent | ControlEvent | EmoteChatEvent | EnvelopeEvent | GoalUpdateEvent | ImDeleteEvent | LikeEvent | RoomUserSeqEvent | SocialEvent | SubscribeEvent | RankUpdateEvent | JoinEvent | PollEvent | QuestionNewEvent | RankTextEvent | HourlyRankEvent | LinkMicArmiesEvent | LinkMicBattleEvent | LinkMicFanTicketMethodEvent | LinkMicMethodEvent | LiveIntroEvent | UnauthorizedMemberEvent | MessageDetectEvent | OecLiveShoppingEvent | RoomPinEvent | SystemEvent | LinkEvent | LinkLayerEvent], f: EventHandler | None = None) → Handler | Callable[[Handler], Handler]

Decorator that can be used to register a Python function as an event listener

Parameters:

• event – The event to listen to

• f – The function to handle the event

Returns:

The wrapped function as a generated pyee.Handler object

classmethod parse_unique_id(unique_id: str) → str

Parse unique ID from a generic string

Parameters:

unique_id – The unique_id to parse

Returns:

The parsed unique_id

property room_id: str | None

The room ID the user is currently connected to

Returns:

Room ID or None

property room_info: dict | None

Information about the room if fetch_room_info=True when starting the client (e.g. with client.run)

Returns:

Dictionary of room info

run(**kwargs) → Task

Start a thread-blocking connection to TikTokLive

Parameters:

kwargs – Kwargs to pass to start

Returns:

The task, once it’s finished

async start(*, process_connect_events: bool = True, fetch_room_info: bool = True, fetch_gift_info: bool = False, room_id: int | None = None) → Task

Create a non-blocking connection to TikTok LIVE and return the task

Parameters:

• process_connect_events – Whether to process initial events sent on room join

• fetch_room_info – Whether to fetch room info on join

• fetch_gift_info – Whether to fetch gift info on join

• room_id – An override to the room ID to connect directly to the livestream and skip scraping the live. Useful when trying to scale, as scraping the HTML can result in TikTok blocks.

Returns:

Task containing the heartbeat of the client

property unique_id: str

The cleaned unique-id parameter passed to the client

property web: TikTokWebClient

The HTTP client that this client uses for requests

Returns:

A copy of the TikTokWebClient

TikTokLive.client.errors module

exception TikTokLive.client.errors.AlreadyConnectedError

Bases: RuntimeError

Thrown when attempting to connect to a user that is already connected to

exception TikTokLive.client.errors.InitialCursorMissingError

Bases: RuntimeError

Thrown when the cursor for connecting to TikTok is missing (blocked)

exception TikTokLive.client.errors.UserOfflineError

Bases: RuntimeError

Thrown when the requested streamer to watch is offline

exception TikTokLive.client.errors.WebsocketURLMissingError

Bases: RuntimeError

Thrown when the websocket URL to connect to TikTok is missing (blocked)

TikTokLive.client.logger module

class TikTokLive.client.logger.LogLevel(value)

Bases: Enum

The level to be used with the python logging module

CRITICAL = 50

DEBUG = 10

ERROR = 40

INFO = 20

NOTSET = 0

WARNING = 30

class TikTokLive.client.logger.TikTokLiveLogHandler(stream: Any | None = None, formatter: Formatter | None = None)

Bases: StreamHandler

A custom logger handler for TikTokLive

FORMAT: str = '[%(name)s] %(levelname)s from %(stack)s:%(lineno)d — %(message)s'

LOGGER: Logger | None = None

LOGGER_NAME: str = 'TikTokLive'

SPACING: Dict[int, int] = {10: 0, 20: 1, 30: 1, 40: 0}

TIME_FORMAT: str = '%H:%M:%S'

emit(record: LogRecord) → None

Handle emitting from the logger

Parameters:

record – The record to emit from

Returns:

None

classmethod format_path(record: LogRecord) → str

Take a path from a stacktrace and convert it into compressed form similar to SpringBoot.

Parameters:

record – The record to read

Returns:

The formatted path in dot-format

classmethod get_logger(level: LogLevel | None = None, stream: Any | None = None) → Logger

Create a logger or retrieve the existing one

Parameters:

• stream – Where to stream to

• level – The level to log above

Returns:

Instance of new logger

Module contents