TikTokLive.client package
Submodules
TikTokLive.client.base module
- class TikTokLive.client.base.BaseClient(unique_id: str, loop: Optional[AbstractEventLoop] = None, client_params: Optional[dict] = None, headers: Optional[dict] = None, timeout_ms: Optional[int] = None, ping_interval_ms: int = 1000, process_initial_data: bool = True, enable_extended_gift_info: bool = True, trust_env: bool = False, proxies: Optional[Dict[str, str]] = None, lang: Optional[str] = 'en-US', fetch_room_info_on_connect: bool = True, websocket_enabled: bool = True, websocket_timeout_ms: int = 15000)
Bases:
AsyncIOEventEmitter
Base client responsible for long polling to the TikTok Webcast API
- property available_gifts: Dict[int, ExtendedGift]
Available gift information for live room
- Returns:
Gift info
- property connected: bool
Whether the client is connected
- Returns:
Result
- download(path: str, duration: Optional[int] = None, verbose: bool = True, loglevel: str = 'error', global_options: Set[str] = {}, inputs: Dict[str, str] = {}, outputs: Dict[str, str] = {}) None
Start downloading the user’s livestream video for a given duration, NON-BLOCKING via Python Threading
- Parameters:
loglevel – Set the FFmpeg log level
outputs – Pass custom params to FFmpeg outputs
inputs – Pass custom params to FFmpeg inputs
global_options – Pass custom params to FFmpeg global options
path – The path to download the livestream video to
duration – If duration is None or less than 1, download will go forever
verbose – Whether to log info about the download in console
- Returns:
None
- Raises:
AlreadyDownloadingStream if already downloading and attempting to start a second download
- async get_proxies() Optional[Dict[str, str]]
Get the current proxies being used in HTTP requests
- Returns:
The current proxies in use
- async retrieve_available_gifts() Optional[Dict[int, ExtendedGift]]
Retrieve available gifts from Webcast API
- Returns:
None
- async retrieve_room_info() Optional[dict]
Method to retrieve room information
- Returns:
Dictionary containing all room info
- property room_id: Optional[int]
Room ID if the connection was successful
- Returns:
Room’s ID
- property room_info: Optional[dict]
Room info dict if the connection was successful
- Returns:
Room Info Dict
- run(session_id: Optional[str] = None) None
Run client while blocking main thread
- Returns:
None
- async send_message(text: str, session_id: Optional[str] = None) Optional[str]
Send a message to the TikTok Live Chat
- Parameters:
text – The message you want to send to the chat
session_id – The Session ID (If you’ve already supplied one, you don’t need to)
- Returns:
The response from the webcast API
- async set_proxies(proxies: Optional[Dict[str, str]]) None
Set the proxies to be used by the HTTP client (Not Websockets)
- Parameters:
proxies – The proxies to use in HTTP requests
- Returns:
None
- async start(session_id: Optional[str] = None) Optional[str]
Start the client without blocking the main thread
- Returns:
Room ID that was connected to
- stop() None
Stop the client safely
- Returns:
None
- stop_download() None
Stop downloading a livestream if currently downloading
- Returns:
None
- Raises:
NotDownloadingStream – Raised if trying to stop when not downloading and
DownloadProcessNotFound – Raised if stopping before the ffmpeg process has opened
- property unique_id: str
Unique ID of the streamer
- Returns:
Their unique ID
- property viewer_count: Optional[int]
Return viewer count of user
- Returns:
Viewer count
TikTokLive.client.client module
- class TikTokLive.client.client.TikTokLiveClient(unique_id: str, debug: bool = False, **options)
Bases:
BaseClient
TikTokLive Client responsible for emitting events asynchronously
TikTokLive.client.config module
- TikTokLive.client.config.DEFAULT_CLIENT_PARAMS: Dict[str, Union[int, bool, str]] = {'aid': 1988, 'app_language': 'en-US', 'app_name': 'tiktok_web', 'browser_language': 'en', 'browser_name': 'Mozilla', 'browser_online': True, 'browser_platform': 'Win32', 'browser_version': '5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/102.0.5005.63 Safari/537.36', 'cookie_enabled': True, 'cursor': '', 'device_platform': 'web', 'did_rule': 3, 'fetch_rule': 1, 'focus_state': True, 'from_page': 'user', 'history_len': 4, 'identity': 'audience', 'internal_ext': '', 'is_fullscreen': False, 'is_page_visible': True, 'last_rtt': 0, 'live_id': 12, 'msToken': '', 'referer': 'https://www.tiktok.com/', 'resp_content_type': 'protobuf', 'root_referer': 'https://www.tiktok.com/', 'screen_height': 1152, 'screen_width': 2048, 'tz_name': 'Europe/Berlin', 'update_version_code': '1.3.0', 'version_code': 180800, 'webcast_sdk_version': '1.3.0'}
Default HTTP client headers to include in requests to the Webcast API & Websocket Server
- TikTokLive.client.config.DEFAULT_REQUEST_HEADERS: Dict[str, str] = {'Accept': 'text/html,application/json,application/protobuf', 'Accept-Encoding': 'gzip, deflate', 'Accept-Language': 'en-US,en;q=0.9', 'Cache-Control': 'max-age=0', 'Connection': 'keep-alive', 'Origin': 'https://www.tiktok.com', 'Referer': 'https://www.tiktok.com/', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/102.0.5005.63 Safari/537.36'}
The URL of the TikTok Webapp
- TikTokLive.client.config.TIKTOK_URL_WEB: str = 'https://www.tiktok.com/'
The URL of the Webcast API
- TikTokLive.client.config.TIKTOK_URL_WEBCAST: str = 'https://webcast.tiktok.com/webcast/'
The URL of the Webcast External Signing API
TikTokLive.client.httpx module
- class TikTokLive.client.httpx.TikTokHTTPClient(headers: Optional[Dict[str, str]] = None, timeout_ms: Optional[int] = None, proxies: Optional[Dict[str, str]] = None, trust_env: bool = True, params: Optional[Dict[str, str]] = {})
Bases:
object
Client for making HTTP requests to TikTok’s Webcast API
- async get_deserialized_object_from_webcast_api(path: str, params: dict, schema: str, sign_url: bool = False) dict
Retrieve and deserialize an object from the Webcast API
- Parameters:
sign_url – Whether to sign the URL (if it’s an authenticated request)
path – Webcast path
params – Parameters to encode into URL
schema – Proto schema to decode from
- Returns:
Deserialized data from API in dictionary format
- Raises:
httpx.TimeoutException
- async get_json_object_from_webcast_api(path: str, params: dict) dict
Retrieve JSON formatted data from the Webcast API
- Parameters:
path – Webcast path
params – Parameters to encode into URL
- Returns:
JSON data from Webcast API
- Raises:
httpx.TimeoutException
- async get_livestream_page_html(unique_id: str) str
Get livestream page HTML given a unique id
- Parameters:
unique_id – Unique ID of the streamer
- Returns:
HTML string containing page data
- Raises:
httpx.TimeoutException
- async post_json_to_webcast_api(path: str, params: dict, json: Optional[dict] = None, sign_url: bool = False) dict
Post JSON to the Webcast API
- Parameters:
sign_url – Whether to sign the URL (if it’s an authenticated request)
path – Path to POST
params – URLEncoded Params
json – JSON Data
- Returns:
Result from the Webcast API POST request
- Raises:
httpx.TimeoutException
- classmethod update_url(url: str, params: dict) str
Update a URL with given parameters by breaking it into components, adding new ones, and rebuilding it
- Parameters:
url – The URL we are updating
params – The parameters to update it with
- Returns:
The updated URL