TikTokLive.client package

Subpackages

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 | None = None, ws_kwargs: dict | None = None)

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(callback: Callable[[None], None] | Callable[[None], Coroutine[None, None, None]] | Coroutine[None, None, None] | None = None, **kwargs) → Task

Start a future-blocking connection to TikTokLive

Parameters:

  • callback – A callback function to run when connected

  • 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

async handle_custom_event(response: WebcastResponseMessage, event: 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) → WebsocketResponseEvent | UnknownEvent | ConnectEvent | FollowEvent | ShareEvent | LiveEndEvent | LivePauseEvent | LiveUnpauseEvent | DisconnectEvent | None

Extract CustomEvent events from existing ProtoEvent events

Parameters:

  • response – The WebcastResponseMessage to parse for the custom event

  • event – The ProtoEvent to parse for the custom event

Returns:

The event, if one exists

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: int | 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 = False, fetch_gift_info: bool = False, fetch_live_check: bool = True, 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

  • fetch_live_check – Whether to check if the user is live (you almost ALWAYS want this enabled)

  • 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.AgeRestrictedError

Bases: RuntimeError

Thrown when a LIVE is age restricted. Pass sessionid to bypass.

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.UserNotFoundError(unique_id: str, *args)

Bases: RuntimeError

Thrown when the request to check if a user is live fails because a user has no livestream account (e.g. <1000 followers)

exception TikTokLive.client.errors.UserOfflineError

Bases: RuntimeError

Thrown when the requested streamer to watch is offline

exception TikTokLive.client.errors.WebcastBlocked200Error

Bases: RuntimeError

Thrown when the webcast is blocked by TikTok with a 200 status code (detected)

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, names=None, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

The level to be used with the python logging module

CRITICAL = 50
DEBUG = 10
ERROR = 40
INFO = 20
NOTSET = 0
WARNING = 30
property value: int

Return the enum item value

Returns:

Value recast (correctly) as an int

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