diff --git a/.sphinx/TikTokLive.client.rst b/.sphinx/TikTokLive.client.rst
index 08e2d24..e6a5c1b 100644
--- a/.sphinx/TikTokLive.client.rst
+++ b/.sphinx/TikTokLive.client.rst
@@ -20,6 +20,14 @@ TikTokLive.client.client module
:undoc-members:
:show-inheritance:
+TikTokLive.client.config module
+-------------------------------
+
+.. automodule:: TikTokLive.client.config
+ :members:
+ :undoc-members:
+ :show-inheritance:
+
TikTokLive.client.http module
-----------------------------
@@ -28,10 +36,10 @@ TikTokLive.client.http module
:undoc-members:
:show-inheritance:
-TikTokLive.client.proxy module
-------------------------------
+TikTokLive.client.websocket module
+----------------------------------
-.. automodule:: TikTokLive.client.proxy
+.. automodule:: TikTokLive.client.websocket
:members:
:undoc-members:
:show-inheritance:
diff --git a/.sphinx/_build/doctrees/README.doctree b/.sphinx/_build/doctrees/README.doctree
index fa04cc1..b6f8f8e 100644
Binary files a/.sphinx/_build/doctrees/README.doctree and b/.sphinx/_build/doctrees/README.doctree differ
diff --git a/.sphinx/_build/doctrees/TikTokLive.client.doctree b/.sphinx/_build/doctrees/TikTokLive.client.doctree
index dfbd3f3..be40dc3 100644
Binary files a/.sphinx/_build/doctrees/TikTokLive.client.doctree and b/.sphinx/_build/doctrees/TikTokLive.client.doctree differ
diff --git a/.sphinx/_build/doctrees/TikTokLive.doctree b/.sphinx/_build/doctrees/TikTokLive.doctree
index 57d29ce..478a94c 100644
Binary files a/.sphinx/_build/doctrees/TikTokLive.doctree and b/.sphinx/_build/doctrees/TikTokLive.doctree differ
diff --git a/.sphinx/_build/doctrees/TikTokLive.proto.doctree b/.sphinx/_build/doctrees/TikTokLive.proto.doctree
index c5c7a3b..11e306d 100644
Binary files a/.sphinx/_build/doctrees/TikTokLive.proto.doctree and b/.sphinx/_build/doctrees/TikTokLive.proto.doctree differ
diff --git a/.sphinx/_build/doctrees/TikTokLive.types.doctree b/.sphinx/_build/doctrees/TikTokLive.types.doctree
index 6e3a44b..dceae0b 100644
Binary files a/.sphinx/_build/doctrees/TikTokLive.types.doctree and b/.sphinx/_build/doctrees/TikTokLive.types.doctree differ
diff --git a/.sphinx/_build/doctrees/environment.pickle b/.sphinx/_build/doctrees/environment.pickle
index 1d90d56..344f2d9 100644
Binary files a/.sphinx/_build/doctrees/environment.pickle and b/.sphinx/_build/doctrees/environment.pickle differ
diff --git a/.sphinx/_build/doctrees/index.doctree b/.sphinx/_build/doctrees/index.doctree
index d241107..6f93b28 100644
Binary files a/.sphinx/_build/doctrees/index.doctree and b/.sphinx/_build/doctrees/index.doctree differ
diff --git a/.sphinx/_build/doctrees/modules.doctree b/.sphinx/_build/doctrees/modules.doctree
index 258b706..2782dc2 100644
Binary files a/.sphinx/_build/doctrees/modules.doctree and b/.sphinx/_build/doctrees/modules.doctree differ
diff --git a/README.md b/README.md
index 2afdc7c..2745b7b 100644
--- a/README.md
+++ b/README.md
@@ -1,19 +1,19 @@
-TikTokLive (Note: Issue with connecting known, being worked on)
+TikTokLive
==================
A python library to connect to and read events from TikTok's LIVE service
[](https://www.linkedin.com/in/isaac-kogan-5a45b9193/ )
[](http://hits.dwyl.com/isaackogan/TikTokLive)
-
-
-
-
+
+
+
+
[](https://discord.gg/e2XwPNTBBr)
-A python library to receive and decode livestream events such as comments and gifts in real-time from TikTok's LIVE service by connecting to TikTok's internal WebCast push service. This library includes a wrapper that
-connects to the WebCast service using only a user's `unique_id` and allows you to join your livestream as well as that of other streamers. No credentials are required to use TikTokLive.
+A python library to receive and decode livestream events such as comments and gifts in real-time from TikTok's LIVE service by connecting to TikTok's internal Webcast push service. This library includes a wrapper that
+connects to the Webcast service using only a user's `unique_id` and allows you to join your livestream as well as that of other streamers. No credentials are required to use TikTokLive.
This library a Python implementation of the Javascript
[TikTok-Live-Connector](https://github.com/zerodytrash/TikTok-Live-Connector)
@@ -27,45 +27,40 @@ Join the [support discord](https://discord.gg/e2XwPNTBBr) and visit the `#suppor
**Primary Information**
-- [Documentation](https://tiktoklive.isaackogan.com)
+- [Documentation](https://isaackogan.github.io/TikTokLive/)
- [Contributors](#contributors)
- [License](#license)
-- [Thermal Printing](#-thermal-printing-library-)
+- [Thermal Printing](#thermal-printing-library-for-sale-)
**Resources & Guides**
-1. [David's Intro Tutorial](#intro-tutorial)
+1. [David's Intro Tutorial](#tiktoklive-intro-tutorial)
2. [Getting Started](#getting-started)
3. [Params & Options](#Params-&-Options)
4. [Client Methods](#Methods)
-5. [TikTok Events](#Events)
-6. [Usage Examples](https://github.com/davidteather/TikTok-Api/tree/master/examples)
+5. [Client Attributes](#Attributes)
+6. [TikTok Events](#Events)
+7. [Usage Examples](https://github.com/davidteather/TikTok-Api/tree/master/examples)
-## 💲🖨 Thermal Printing Library 🖨💲
+## Thermal Printing Library for Sale 🖨
-Thermal printing is a very recent, very exciting trend on TikTok.
+Print text, images, text-to-speech, play sounds, and much more. It's just a one-time, life-time purchase, including future support and updates.
-I developed an all-encompassing, multithreaded thermal printing program that does *everything*
-you could ever want with Thermal Printing. It even has its own [YouTube Tutorial](https://www.youtube.com/watch?v=NeapS5Jn_oo) and comes
-with [pre-made examples](https://github.com/isaackogan/TikTokPrinter/tree/master/examples) if your coding ability isn't very strong!
+Install is easy, takes about 15 minutes. Plug your printer in, download the drivers, and run a command in terminal.
-Print text, images, text-to-speech, play sounds, and much more. There is no subscription unlike virtual printer services. It's just a one-time, life-time purchase.
-
-It's so easy, it can be installed in **one** command through pip. It's not just a "one-off" purchase, either. As the project is updated, you will have access to **every new release** as new features are added.
+As the project is updated, you will have access to **every new release** as new features are added.
Here's a sample of what you can do with this library in less than 30 lines of code:
+It even has its own [YouTube Tutorial](https://www.youtube.com/watch?v=NeapS5Jn_oo) and comes with [pre-made examples](https://github.com/isaackogan/TikTokPrinter/tree/master/examples) if your coding ability isn't very
+strong!
[](https://github.com/isaackogan/TikTokPrinter)
### How to Purchase
-First, read more about it on the public [TikTokPrinter](https://github.com/isaackogan/TikTokPrinter) GitHub page.
-
-Then, to buy this library, create a ticket in the `#tickets` channel in https://discord.gg/4Mbw58w5Qx.
+To buy this library, create a ticket in the `#tickets` channel in https://discord.gg/4Mbw58w5Qx.
-Type "Printer Magic" in the ticket to get started with your purchase.
-
-## Intro Tutorial
+## TikTokLive Intro Tutorial
I cannot recommend this tutorial enough for people trying to get started. It is succinct, informative and easy to understand, created by [David Teather](https://github.com/davidteather), the creator of the
Python [TikTok-Api](https://github.com/davidteather/TikTok-Api) package. Click the thumbnail to warp.
@@ -132,17 +127,9 @@ from TikTokLive import TikTokLiveClient
client: TikTokLiveClient = TikTokLiveClient(
unique_id="@oldskoldj", **(
{
- # Whether to process initial data (cached chats, etc.)
- "process_initial_data": True,
-
- # Connect info (viewers, stream status, etc.)
- "fetch_room_info_on_connect": True,
- # Whether to get extended gift info (Image URLs, etc.)
- "enable_extended_gift_info": True,
-
- # How frequently to poll Webcast API
- "polling_interval_ms": 1000,
+ # Custom Asyncio event loop
+ "loop": None,
# Custom Client params
"client_params": {},
@@ -153,23 +140,39 @@ client: TikTokLiveClient = TikTokLiveClient(
# Custom timeout for Webcast API requests
"timeout_ms": 1000,
- # Custom Asyncio event loop
- "loop": None,
+ # How frequently to poll Webcast API
+ "ping_interval_ms": 1000,
+
+ # Whether to process initial data (cached chats, etc.)
+ "process_initial_data": True,
+
+ # Whether to get extended gift info (Image URLs, etc.)
+ "enable_extended_gift_info": True,
- # Whether to trust environment variables that provide proxies to be used in aiohttp requests
+ # Whether to trust environment variables that provide proxies to be used in http requests
"trust_env": False,
- # A ProxyContainer object for proxied requests
- "proxy_container": None,
+ # A dict object for proxies requests
+ "proxies": {
+ "http://": "http://username:password@localhost:8030",
+ "https://": "http://420.69.420:8031",
+ },
# Set the language for Webcast responses (Changes extended_gift's language)
- "lang": "en-US"
+ "lang": "en-US",
+
+ # Connect info (viewers, stream status, etc.)
+ "fetch_room_info_on_connect": True,
+
+ # Whether to allow Websocket connections
+ "websocket_enabled": False
}
)
)
-client.run()
+if __name__ == "__main__":
+ client.run()
```
## Methods
@@ -184,12 +187,22 @@ A `TikTokLiveClient` object contains the following methods.
| retrieve_room_info | Gets the current room info from TikTok API |
| retrieve_available_gifts | Retrieves a list of the available gifts for the room and adds it to the `extended_gift` attribute of the `Gift` object on the `gift` event, when enabled. |
| add_listener | Adds an *asynchronous* listener function (or, you can decorate a function with `@client.on()`) and takes two parameters, an event name and the payload, an AbstractEvent ||
-| add_proxies | Add proxies to the current list of proxies with a valid aiohttp proxy-url |
-| get_proxies | Get the current list of proxies by proxy-url |
-| remove_proxies | Remove proxies from the current list of proxies by proxy-url |
-| set_proxies_enabled | Set whether or not proxies are enabled (disabled by default) |
| download | Start downloading the livestream video for a given duration or until stopped via the `stop_download` method |
| stop_download | Stop downloading the livestream video if currently downloading, otherwise throws an error |
+| send_message | Send a message to the TikTok LIVE chat using session cookies |
+| set_proxies | Set proxies to be used in HTTP requests (excludes the Websocket connection) |
+| get_proxies | Get the current proxies being used for requests |
+
+## Attributes
+
+| Attribute Name | Description |
+|-----------------|-----------------------------------------------------------------------------------------|
+| viewer_count | The number of people currently watching the livestream broadcast |
+| room_id | The ID of the livestream room the client is currently connected to |
+| room_info | Information about the given livestream room |
+| unique_id | The TikTok username of the person whose livestream the client is currently connected to |
+| connected | Whether the client is currently connected to a livestream |
+| available_gifts | A dictionary containing K:V pairs of `Dict[int, ExtendedGift]` | |
## Events
@@ -399,11 +412,11 @@ async def on_connect(event: UnknownEvent):
### `error`
-Triggered when there is an error in the client or in error handlers.
+Triggered when there is an error in the client or error handlers.
If this handler is not present in the code, an internal default handler will log errors in the console. If a handler is added, all error handling (including logging) is up to the individual.
-**Warning:** If you listen for the error event and do not log errors, you will not see when an error occurs.
+**Warning:** If you listen for the error event and do not log errors, you will not see when an error occurs. This is because listening to the error event causes the default one to be overriden/turned off.
```python
@@ -422,7 +435,7 @@ async def on_connect(error: Exception):
## Contributors
* **Isaac Kogan** - *Initial work & primary maintainer* - [isaackogan](https://github.com/isaackogan)
-* **Zerody** - *Reverse-Engineering & README.md file* - [Zerody](https://github.com/zerodytrash/)
+* **Zerody** - *Reverse-Engineering & Support* - [Zerody](https://github.com/zerodytrash/)
* **Davincible** - *Reverse-Engineering Stream Downloads* - [davincible](https://github.com/davincible)
* **David Teather** - *TikTokLive Introduction Tutorial* - [davidteather](https://github.com/davidteather)
diff --git a/TikTokLive/client/base.py b/TikTokLive/client/base.py
index 227da0a..38d1ffe 100644
--- a/TikTokLive/client/base.py
+++ b/TikTokLive/client/base.py
@@ -11,15 +11,17 @@
from dacite import from_dict
from ffmpy import FFmpeg, FFRuntimeError
+from pyee import AsyncIOEventEmitter
+from TikTokLive.client import config
from TikTokLive.client.http import TikTokHTTPClient
-from TikTokLive.client.proxy import ProxyContainer
+from TikTokLive.client.websocket import WebcastWebsocket
from TikTokLive.types import AlreadyConnecting, AlreadyConnected, LiveNotFound, FailedConnection, ExtendedGift, InvalidSessionId, ChatMessageSendFailure, ChatMessageRepeat, FailedFetchRoomInfo, FailedFetchGifts, \
- FailedRoomPolling, FFmpegWrapper, AlreadyDownloadingStream, DownloadProcessNotFound, NotDownloadingStream
+ FailedRoomPolling, FFmpegWrapper, AlreadyDownloadingStream, DownloadProcessNotFound, NotDownloadingStream, InitialCursorMissing
from TikTokLive.utils import validate_and_normalize_unique_id, get_room_id_from_main_page_html
-class BaseClient:
+class BaseClient(AsyncIOEventEmitter):
"""
Base client responsible for long polling to the TikTok Webcast API
@@ -32,13 +34,14 @@ def __init__(
client_params: Optional[dict] = None,
headers: Optional[dict] = None,
timeout_ms: Optional[int] = None,
- polling_interval_ms: int = 1000,
+ ping_interval_ms: int = 1000,
process_initial_data: bool = True,
- fetch_room_info_on_connect: bool = True,
enable_extended_gift_info: bool = True,
trust_env: bool = False,
- proxy_container: Optional[ProxyContainer] = None,
- lang: Optional[str] = "en-US"
+ proxies: Optional[Dict[str, str]] = None,
+ lang: Optional[str] = "en-US",
+ fetch_room_info_on_connect: bool = True,
+ websocket_enabled: bool = True,
):
"""
Initialize the base client
@@ -48,15 +51,17 @@ def __init__(
:param client_params: Additional client parameters to include when making requests to the Webcast API
:param headers: Additional headers to include when making requests to the Webcast API
:param timeout_ms: The timeout (in ms) for requests made to the Webcast API
- :param polling_interval_ms: The interval between requests made to the Webcast API
+ :param ping_interval_ms: The interval between requests made to the Webcast API for both Websockets and Long Polling
:param process_initial_data: Whether to process the initial data (including cached chats)
- :param fetch_room_info_on_connect: Whether to fetch room info (check if everything is kosher) on connect
:param enable_extended_gift_info: Whether to retrieve extended gift info including its icon & other important things
:param trust_env: Whether to trust environment variables that provide proxies to be used in aiohttp requests
- :param proxy_container: A proxy container that allows you to submit an unlimited # of proxies for rotation
+ :param proxies: Enable proxied requests by turning on forwarding for the HTTPX "proxies" argument. Websocket connections will NOT be proxied
:param lang: Change the language. Payloads *will* be in English, but this will change stuff like the extended_gift Gift attribute to the desired language!
+ :param fetch_room_info_on_connect: Whether to fetch room info on connect. If disabled, you might attempt to connect to a closed livestream
+ :param websocket_enabled: Whether to use websockets or rely on purely long polling
"""
+ AsyncIOEventEmitter.__init__(self)
# Get Event Loop
if isinstance(loop, AbstractEventLoop):
@@ -77,19 +82,31 @@ def __init__(
self.__connecting: bool = False
self.__connected: bool = False
self.__session_id: Optional[str] = None
+ self.__is_ws_upgrade_done: bool = False
+ self.__websocket_enabled: bool = websocket_enabled
# Change Language
- TikTokHTTPClient.DEFAULT_CLIENT_PARAMS["app_language"] = lang
- TikTokHTTPClient.DEFAULT_CLIENT_PARAMS["webcast_language"] = lang
+ config.DEFAULT_CLIENT_PARAMS["app_language"] = lang
+ config.DEFAULT_CLIENT_PARAMS["webcast_language"] = lang
# Protected Attributes
- self._client_params: dict = {**TikTokHTTPClient.DEFAULT_CLIENT_PARAMS, **(client_params if isinstance(client_params, dict) else dict())}
- self._http: TikTokHTTPClient = TikTokHTTPClient(headers if headers is not None else dict(), timeout_ms=timeout_ms, proxy_container=proxy_container, trust_env=trust_env)
- self._polling_interval_ms: int = polling_interval_ms
+
+ self._http: TikTokHTTPClient = TikTokHTTPClient(
+ headers=headers if headers is not None else dict(),
+ timeout_ms=timeout_ms,
+ proxies=proxies,
+ trust_env=trust_env,
+ params={**config.DEFAULT_CLIENT_PARAMS, **(client_params if isinstance(client_params, dict) else dict())}
+ )
+ self._ping_interval_ms: int = ping_interval_ms
self._process_initial_data: bool = process_initial_data
- self._fetch_room_info_on_connect: bool = fetch_room_info_on_connect
self._enable_extended_gift_info: bool = enable_extended_gift_info
+ self._fetch_room_info_on_connect: bool = fetch_room_info_on_connect
self._download: Optional[FFmpegWrapper] = None
+ self._socket: Optional[WebcastWebsocket] = None
+
+ # Listeners
+ self.add_listener("websocket", self._handle_webcast_messages)
async def _on_error(self, original: Exception, append: Optional[Exception]) -> None:
"""
@@ -115,10 +132,10 @@ async def __fetch_room_id(self) -> Optional[str]:
try:
html: str = await self._http.get_livestream_page_html(self.__unique_id)
self.__room_id = get_room_id_from_main_page_html(html)
- self._client_params["room_id"] = self.__room_id
+ self._http.params["room_id"] = self.__room_id
return self.__room_id
except Exception as ex:
- await self._on_error(ex, FailedFetchRoomInfo("Failed to fetch room id from WebCast, see stacktrace for more info."))
+ await self._on_error(ex, FailedFetchRoomInfo("Failed to fetch room id from Webcast, see stacktrace for more info."))
return None
async def __fetch_room_info(self) -> Optional[dict]:
@@ -130,11 +147,11 @@ async def __fetch_room_info(self) -> Optional[dict]:
"""
try:
- response = await self._http.get_json_object_from_webcast_api("room/info/", self._client_params)
+ response = await self._http.get_json_object_from_webcast_api("room/info/", self._http.params)
self.__room_info = response
return self.__room_info
except Exception as ex:
- await self._on_error(ex, FailedFetchRoomInfo("Failed to fetch room info from WebCast, see stacktrace for more info."))
+ await self._on_error(ex, FailedFetchRoomInfo("Failed to fetch room info from Webcast, see stacktrace for more info."))
return None
async def __fetch_available_gifts(self) -> Optional[Dict[int, ExtendedGift]]:
@@ -146,7 +163,7 @@ async def __fetch_available_gifts(self) -> Optional[Dict[int, ExtendedGift]]:
"""
try:
- response = await self._http.get_json_object_from_webcast_api("gift/list/", self._client_params)
+ response = await self._http.get_json_object_from_webcast_api("gift/list/", self._http.params)
gifts: Optional[List] = response.get("gifts")
if isinstance(gifts, list):
@@ -159,7 +176,7 @@ async def __fetch_available_gifts(self) -> Optional[Dict[int, ExtendedGift]]:
return self.__available_gifts
except Exception as ex:
- await self._on_error(ex, FailedFetchGifts("Failed to fetch gift data from WebCast, see stacktrace for more info."))
+ await self._on_error(ex, FailedFetchGifts("Failed to fetch gift data from Webcast, see stacktrace for more info."))
return None
async def __fetch_room_polling(self) -> None:
@@ -171,13 +188,13 @@ async def __fetch_room_polling(self) -> None:
"""
self.__is_polling_enabled = True
- polling_interval: int = int(self._polling_interval_ms / 1000)
+ polling_interval: float = self._ping_interval_ms / 1000
while self.__is_polling_enabled:
try:
await self.__fetch_room_data()
except Exception as ex:
- await self._on_error(ex, FailedRoomPolling("Failed to retrieve events from WebCast, see stacktrace for more info."))
+ await self._on_error(ex, FailedRoomPolling("Failed to retrieve events from Webcast, see stacktrace for more info."))
await asyncio.sleep(polling_interval)
@@ -190,18 +207,53 @@ async def __fetch_room_data(self, is_initial: bool = False) -> None:
"""
- webcast_response = await self._http.get_deserialized_object_from_webcast_api("im/fetch/", self._client_params, "WebcastResponse")
- _last_cursor, _next_cursor = self._client_params["cursor"], webcast_response.get("cursor")
- self._client_params["cursor"] = _last_cursor if _next_cursor == "0" else _next_cursor
+ # Fetch from polling api
+ webcast_response = await self._http.get_deserialized_object_from_webcast_api("im/fetch/", self._http.params, "WebcastResponse", is_initial)
+ _last_cursor, _next_cursor = self._http.params["cursor"], webcast_response.get("cursor")
+ self._http.params["cursor"] = _last_cursor if _next_cursor == "0" else _next_cursor
+ # Add param if given
if webcast_response.get("internalExt"):
- self._client_params["internal_ext"] = webcast_response["internalExt"]
+ self._http.params["internal_ext"] = webcast_response["internalExt"]
- if is_initial and not self._process_initial_data:
- return
+ if is_initial:
+ if not webcast_response.get("cursor"):
+ raise InitialCursorMissing("Missing cursor in initial fetch response.")
+
+ # If a WebSocket is offered, upgrade
+ if bool(webcast_response.get("wsUrl")) and bool(webcast_response.get("wsParam")) and self.__websocket_enabled:
+ await self.__try_websocket_upgrade(webcast_response)
+
+ # Process initial data if requested
+ if not self._process_initial_data:
+ return
await self._handle_webcast_messages(webcast_response)
+ async def __try_websocket_upgrade(self, webcast_response) -> WebcastWebsocket:
+ """
+ Attempt to upgrade the connection to a websocket instead
+
+ :param webcast_response: The initial webcast response including the wsParam and wsUrl items
+ :return: The websocket, if one is produced
+
+ """
+
+ socket: WebcastWebsocket = WebcastWebsocket(
+ client=self,
+ ws_url=webcast_response.get("wsUrl"),
+ cookies=self._http.client.cookies,
+ client_params=self._http.params,
+ ws_params={"imprp": webcast_response.get("wsParam").get("value")},
+ headers=self._http.headers,
+ loop=self.loop,
+ ping_interval_ms=self._ping_interval_ms
+ )
+
+ self.__is_ws_upgrade_done = await socket.connect()
+ self._socket = socket if self.__is_ws_upgrade_done else None
+ return self._socket
+
async def _handle_webcast_messages(self, webcast_response) -> None:
"""
Handle the parsing of webcast messages, meant to be overridden by superclass
@@ -210,13 +262,14 @@ async def _handle_webcast_messages(self, webcast_response) -> None:
raise NotImplementedError
- async def _connect(self) -> str:
+ async def _connect(self, session_id: str = None) -> str:
"""
- Connect to the Websocket API
+ Connect to the WebcastWebsocket API
:return: The room ID, if connection is successful
"""
+ self.__set_session_id(session_id)
if self.__connecting:
raise AlreadyConnecting()
@@ -235,7 +288,7 @@ async def _connect(self) -> str:
# If offline
if self.__room_info.get("status", 4) == 4:
- raise LiveNotFound()
+ raise LiveNotFound("The requested user is most likely offline.")
# Get extended gift info
if self._enable_extended_gift_info:
@@ -245,8 +298,19 @@ async def _connect(self) -> str:
await self.__fetch_room_data(True)
self.__connected = True
- # Use request polling (Websockets not implemented)
- self.loop.create_task(self.__fetch_room_polling())
+ # If the websocket was not connected for whatever reason
+ if not self.__is_ws_upgrade_done:
+ # Switch to long polling if a session id was provided
+ if self._http.client.cookies.get("sessionid"):
+ self.loop.create_task(self.__fetch_room_polling())
+
+ else:
+ # No more options, fail to connect
+ raise FailedRoomPolling(
+ ("You have disabled websockets, but not included a sessionid for long polling. " if not self.__websocket_enabled else "")
+ + "Long polling is not available: Try adding a sessionid as an argument in start() or run()"
+ )
+
return self.__room_id
except Exception as ex:
@@ -277,11 +341,11 @@ def _disconnect(self) -> None:
self.__room_info: Optional[dict] = None
self.__connecting: Optional[bool] = False
self.__connected: Optional[bool] = False
- self._client_params["cursor"]: str = ""
+ self._http.params["cursor"]: str = ""
async def stop(self) -> None:
"""
- Stop the client
+ Stop the client safely
:return: None
@@ -299,8 +363,7 @@ async def start(self, session_id: Optional[str] = None) -> Optional[str]:
"""
- self.__set_session_id(session_id)
- return await self._connect()
+ return await self._connect(session_id=session_id)
def run(self, session_id: Optional[str] = None) -> None:
"""
@@ -309,9 +372,8 @@ def run(self, session_id: Optional[str] = None) -> None:
:return: None
"""
- self.__set_session_id(session_id)
- self.loop.run_until_complete(self._connect())
+ self.loop.run_until_complete(self._connect(session_id=session_id))
self.loop.run_forever()
def __set_session_id(self, session_id: Optional[str]) -> None:
@@ -325,7 +387,7 @@ def __set_session_id(self, session_id: Optional[str]) -> None:
if session_id:
self.__session_id = session_id
- self._http.cookies["sessionid"] = session_id
+ self._http.client.cookies.set("sessionid", session_id)
async def send_message(self, text: str, session_id: Optional[str] = None) -> Optional[str]:
"""
@@ -333,7 +395,7 @@ async def send_message(self, text: str, session_id: Optional[str] = None) -> Opt
:param text: The message you want to send to the chat
:param session_id: The Session ID (If you've already supplied one, you don't need to)
- :return: None
+ :return: The response from the webcast API
"""
@@ -342,8 +404,8 @@ async def send_message(self, text: str, session_id: Optional[str] = None) -> Opt
if not self.__session_id:
raise InvalidSessionId("Missing Session ID. Please provide your current Session ID to use this feature.")
- params: dict = {**self._client_params, "content": text}
- response: dict = await self._http.post_json_to_webcast_api("room/chat/", params, None)
+ params: dict = {**self._http.params, "content": text}
+ response: dict = await self._http.post_json_to_webcast_api("room/chat/", params, None, sign_url=False)
status_code: Optional[int] = response.get("status_code")
data: Optional[dict] = response.get("data")
@@ -385,114 +447,26 @@ async def retrieve_available_gifts(self) -> Optional[Dict[int, ExtendedGift]]:
return await self.__fetch_available_gifts()
- async def set_proxies_enabled(self, enabled: bool) -> None:
- """
- Set whether to use proxies in requests
-
- :param enabled: Whether proxies are enabled or not
- :return: None
-
- """
-
- self._http.proxy_container.set_enabled(enabled)
-
- async def add_proxies(self, *proxies: str) -> None:
+ async def set_proxies(self, proxies: Optional[Dict[str, str]]) -> None:
"""
- Add proxies to the proxy container for request usage
-
- :param proxies: Proxies for usage
- :return: None
-
- """
-
- for proxy in proxies:
- self._http.proxy_container.proxies.append(proxy)
-
- async def remove_proxies(self, *proxies: str) -> None:
- """
- Remove proxies from the proxy container for request usage
+ Set the proxies to be used by the HTTP client (Not Websockets)
- :param proxies: Proxies to remove
- :raises ValueError: Raises ValueError if proxy is not present
+ :param proxies: The proxies to use in HTTP requests
:return: None
"""
- for proxy in proxies:
- self._http.proxy_container.proxies.remove(proxy)
-
- async def get_proxies(self) -> List[str]:
- """
- Get a list of the current proxies in the proxy container being used for requests
-
- :return: The proxies in the request container
- """
-
- return self._http.proxy_container.proxies
-
- @property
- def viewer_count(self) -> Optional[int]:
- """
- Return viewer count of user
-
- :return: Viewer count
-
- """
- return self._viewer_count
-
- @property
- def room_id(self) -> Optional[int]:
- """
- Room ID if the connection was successful
-
- :return: Room's ID
-
- """
- return self.__room_id
-
- @property
- def room_info(self) -> Optional[dict]:
- """
- Room info dict if the connection was successful
-
- :return: Room Info Dict
-
- """
-
- return self.__room_info
-
- @property
- def unique_id(self) -> str:
- """
- Unique ID of the streamer
-
- :return: Their unique ID
-
- """
-
- return self.__unique_id
+ self._http.proxies = proxies
- @property
- def connected(self) -> bool:
+ async def get_proxies(self) -> Optional[Dict[str, str]]:
"""
- Whether the client is connected
+ Get the current proxies being used in HTTP requests
- :return: Result
+ :return: The current proxies in use
"""
- return self.__connected
-
- @property
- def available_gifts(self) -> Dict[int, ExtendedGift]:
- """
- Available gift information for live room
-
- :return: Gift info
-
- """
-
- return self.__available_gifts
+ return self._http.proxies
def download(
self,
@@ -587,3 +561,67 @@ def stop_download(self) -> None:
f"Stopped the download to path \"{self._download.path}\" on user @{self.unique_id} after "
f"\"{int(datetime.utcnow().timestamp()) - self._download.started_at} seconds\" of downloading"
)
+
+ @property
+ def viewer_count(self) -> Optional[int]:
+ """
+ Return viewer count of user
+
+ :return: Viewer count
+
+ """
+ return self._viewer_count
+
+ @property
+ def room_id(self) -> Optional[int]:
+ """
+ Room ID if the connection was successful
+
+ :return: Room's ID
+
+ """
+ return self.__room_id
+
+ @property
+ def room_info(self) -> Optional[dict]:
+ """
+ Room info dict if the connection was successful
+
+ :return: Room Info Dict
+
+ """
+
+ return self.__room_info
+
+ @property
+ def unique_id(self) -> str:
+ """
+ Unique ID of the streamer
+
+ :return: Their unique ID
+
+ """
+
+ return self.__unique_id
+
+ @property
+ def connected(self) -> bool:
+ """
+ Whether the client is connected
+
+ :return: Result
+
+ """
+
+ return self.__connected
+
+ @property
+ def available_gifts(self) -> Dict[int, ExtendedGift]:
+ """
+ Available gift information for live room
+
+ :return: Gift info
+
+ """
+
+ return self.__available_gifts
diff --git a/TikTokLive/client/client.py b/TikTokLive/client/client.py
index 1bc811f..3edf5e4 100644
--- a/TikTokLive/client/client.py
+++ b/TikTokLive/client/client.py
@@ -3,7 +3,6 @@
from typing import Optional, Type, Callable
from dacite import from_dict
-from pyee import AsyncIOEventEmitter
from .base import BaseClient
from ..proto.utilities import from_dict_plus
@@ -12,7 +11,7 @@
SubscribeEvent, WeeklyRankingEvent, MicBattleEvent, MicArmiesEvent
-class TikTokLiveClient(AsyncIOEventEmitter, BaseClient):
+class TikTokLiveClient(BaseClient):
"""
TikTokLive Client responsible for emitting events asynchronously
@@ -20,6 +19,7 @@ class TikTokLiveClient(AsyncIOEventEmitter, BaseClient):
def __init__(self, unique_id: str, debug: bool = False, **options):
"""
+ Initialize the BaseClient for TikTokLive Webcast tracking
:param unique_id: The unique id of the creator to connect to
:param debug: Debug mode -> Add all events' raw payload to a "debug" event
@@ -28,9 +28,7 @@ def __init__(self, unique_id: str, debug: bool = False, **options):
"""
self.debug_enabled: bool = debug
-
BaseClient.__init__(self, unique_id, **options)
- AsyncIOEventEmitter.__init__(self, self.loop)
async def _on_error(self, original: Exception, append: Optional[Exception]) -> None:
"""
@@ -60,7 +58,7 @@ async def _on_error(self, original: Exception, append: Optional[Exception]) -> N
self._log_error(_exc)
return
- # If connected, has handler
+ # If connected, has handler
self.emit("error", _exc)
@classmethod
@@ -79,13 +77,13 @@ def _log_error(cls, exception: Exception) -> None:
logging.error(traceback.format_exc())
return
- async def _connect(self) -> str:
+ async def _connect(self, session_id: str = None) -> str:
"""
Wrap connection in a connect event
"""
- result: str = await super(TikTokLiveClient, self)._connect()
+ result: str = await super(TikTokLiveClient, self)._connect(session_id=session_id)
if self.connected:
event: ConnectEvent = ConnectEvent()
@@ -122,6 +120,14 @@ async def _handle_webcast_messages(self, webcast_response: dict) -> None:
self.emit("debug", AbstractEvent(data=message))
def __parse_message(self, webcast_message: dict) -> Optional[AbstractEvent]:
+ """
+ Parse a webcast message into an event and return to the caller
+
+ :param webcast_message: The message to parse
+ :return: The parsed object of base-type AbstractEvent
+
+ """
+
event_dict: Optional[dict] = webcast_message.get("event")
# It's a traditional event
diff --git a/TikTokLive/client/config.py b/TikTokLive/client/config.py
new file mode 100644
index 0000000..404836c
--- /dev/null
+++ b/TikTokLive/client/config.py
@@ -0,0 +1,58 @@
+from typing import Dict, Union
+
+"""Default HTTP client parameters to include in requests to the Webcast API & Websocket Server"""
+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": '',
+ "internal_ext": '',
+ "device_platform": 'web',
+ "focus_state": True,
+ "from_page": 'user',
+ "history_len": 4,
+ "is_fullscreen": False,
+ "is_page_visible": True,
+ "did_rule": 3,
+ "fetch_rule": 1,
+ "identity": 'audience',
+ "last_rtt": 0,
+ "live_id": 12,
+ "resp_content_type": 'protobuf',
+ "screen_height": 1152,
+ "screen_width": 2048,
+ "tz_name": 'Europe/Berlin',
+ "referer": 'https://www.tiktok.com/',
+ "root_referer": 'https://www.tiktok.com/',
+ "msToken": '',
+ "version_code": 180800,
+ "webcast_sdk_version": '1.3.0',
+ "update_version_code": '1.3.0',
+}
+
+"""Default HTTP client headers to include in requests to the Webcast API & Websocket Server"""
+DEFAULT_REQUEST_HEADERS: Dict[str, str] = {
+ "Connection": 'keep-alive',
+ 'Cache-Control': 'max-age=0',
+ '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',
+ "Accept": 'text/html,application/json,application/protobuf',
+ "Referer": 'https://www.tiktok.com/',
+ "Origin": 'https://www.tiktok.com',
+ 'Accept-Language': 'en-US,en;q=0.9',
+ 'Accept-Encoding': 'gzip, deflate',
+}
+
+"""The URL of the TikTok Webapp"""
+TIKTOK_URL_WEB: str = 'https://www.tiktok.com/'
+
+"""The URL of the Webcast API"""
+TIKTOK_URL_WEBCAST: str = 'https://webcast.tiktok.com/webcast/'
+
+"""The URL of the Webcast External Signing API"""
+TIKTOK_SIGN_API: str = "https://tiktok.isaackogan.com/"
diff --git a/TikTokLive/client/http.py b/TikTokLive/client/http.py
index 8daa167..b56eb04 100644
--- a/TikTokLive/client/http.py
+++ b/TikTokLive/client/http.py
@@ -1,9 +1,10 @@
+import json as json_parse
import urllib.parse
-from typing import Dict, Union, Optional
+from typing import Dict, Optional
-from aiohttp import ClientSession
+import httpx
-from TikTokLive.client.proxy import ProxyContainer
+from TikTokLive.client import config
from TikTokLive.proto.utilities import deserialize_message
@@ -13,94 +14,114 @@ class TikTokHTTPClient:
"""
- TIKTOK_URL_WEB: str = 'https://www.tiktok.com/'
- TIKTOK_URL_WEBCAST: str = 'https://webcast.tiktok.com/webcast/'
- TIKTOK_HTTP_ORIGIN: str = 'https://www.tiktok.com'
-
- DEFAULT_CLIENT_PARAMS: Dict[str, Union[int, bool, str]] = {
- "aid": 1988, "app_name": 'tiktok_web', "browser_name": 'Mozilla',
- "browser_online": True, "browser_platform": 'Win32', "version_code": 180800,
- "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, "identity": 'audience', "internal_ext": '',
- "last_rtt": 0, "live_id": 12, "resp_content_type": 'protobuf', "screen_height": 1152, "screen_width": 2048, "tz_name": 'Europe/Berlin',
- "browser_language": "en", "priority_region": "US", "region": "US",
- }
-
- DEFAULT_REQUEST_HEADERS: Dict[str, str] = {
- "Connection": 'keep-alive', "Cache-Control": 'max-age=0', "Accept": 'text/html,application/json,application/protobuf',
- "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',
- "Referer": 'https://www.tiktok.com/', "Origin": 'https://www.tiktok.com', "Accept-Language": 'en-US,en;q=0.9', "Accept-Encoding": 'gzip, deflate',
- }
-
- def __init__(self, headers: Optional[Dict[str, str]] = None, timeout_ms: Optional[int] = None, proxy_container: Optional[ProxyContainer] = None, trust_env: bool = True) -> None:
+ def __init__(
+ self,
+ 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]] = dict(),
+ ):
"""
- Initialize HTTP client
+ Initialize HTTP client for TikTok-related requests
:param headers: Headers to use to make HTTP requests
:param timeout_ms: Timeout for HTTP requests
- :param proxy_container: Proxy container to hold proxies for request usage
+ :param proxies: Enable proxied requests by turning on forwarding for the HTTPX "proxies" argument
:param trust_env: Whether to trust the environment when it comes to proxy usage
"""
self.timeout: int = int((timeout_ms if isinstance(timeout_ms, int) else 10000) / 1000)
- self.proxy_container: ProxyContainer = proxy_container if proxy_container is not None else ProxyContainer(enabled=False)
- self.headers: Dict[str, str] = {
- **self.DEFAULT_REQUEST_HEADERS,
- **(headers if isinstance(headers, dict) else dict())
- }
- self.cookies: dict = dict()
+ self.proxies: Optional[Dict[str, str]] = proxies
+ self.headers: Dict[str, str] = {**config.DEFAULT_REQUEST_HEADERS, **(headers if isinstance(headers, dict) else dict())}
+ self.params: dict = params if params else dict()
- async def __aiohttp_get_bytes(self, url: str, params: dict = None) -> bytes:
+ self.trust_env: bool = trust_env
+ self.client = httpx.AsyncClient(trust_env=trust_env)
+ self.__tokens: dict = {}
+
+ @classmethod
+ def update_url(cls, url: str, params: dict) -> str:
"""
- Get bytes from a given URL with parameters
+ Update a URL with given parameters by breaking it into components, adding new ones, and rebuilding it
- :param url: URL to request data from
- :param params: Custom Parameters
- :return: bytearray containing request data
- :raises: asyncio.TimeoutError
+ :param url: The URL we are updating
+ :param params: The parameters to update it with
+ :return: The updated URL
+
+ """
+
+ parsed = list(urllib.parse.urlparse(url))
+ query = {**params, **dict(urllib.parse.parse_qsl(parsed[4]))}
+ parsed[4] = urllib.parse.urlencode(query)
+ return urllib.parse.urlunparse(parsed)
+ async def __get_signed_url(self, url: str) -> str:
"""
- request_url: str = f"{url}?{urllib.parse.urlencode(params if params is not None else dict())}"
+ Sign a URL via external signing agent to authenticate against TikTok's Webcast API.
+ This is an API made *for* this library, NOT by TikTok.
- async with ClientSession() as session:
- async with session.get(request_url, headers=self.headers, timeout=self.timeout, proxy=self.proxy_container.get()) as request:
- return await request.read()
+ :param url: The URL to sign
+ :return: The signed URL
- async def __aiohttp_get_json(self, url: str, params: dict) -> dict:
"""
- Get json (dict form) from a given URL with parameters
+
+ # Get the signed URL
+ response: httpx.Response = await self.client.get(
+ url=f"{config.TIKTOK_SIGN_API}webcast/sign_url?client=ttlive-python&url={urllib.parse.quote(url)}",
+ timeout=self.timeout
+ )
+
+ # Update client information
+ tokens: dict = response.json()
+ self.headers["User-Agent"] = tokens.get("User-Agent")
+ return tokens.get("signedUrl")
+
+ async def __httpx_get_bytes(self, url: str, params: dict = None, sign_url: bool = False) -> bytes:
+ """
+ Get byte data from the Webcast API
+
+ :param url: The URL to request
+ :param params: Parameters to include in the URL
+ :param sign_url: Whether to sign the URL (for authenticated endpoints)
+ :return: The result of the request (a bytearray)
+ :raises: httpx.TimeoutException
+ """
+
+ url: str = self.update_url(url, params if params else dict())
+ response: httpx.Response = await self.client.get(await self.__get_signed_url(url) if sign_url else url, headers=self.headers, timeout=self.timeout)
+ return response.read()
+
+ async def __aiohttp_get_json(self, url: str, params: dict, sign_url: bool = False) -> dict:
+ """
+ Get json (dict) from a given URL with parameters from the Webcast API
:param url: URL to request data from
:param params: Custom Parameters
:return: bytearray containing request data
- :raises: asyncio.TimeoutError
+ :raises: httpx.TimeoutException
"""
- request_url: str = f"{url}?{urllib.parse.urlencode(params if params is not None else dict())}"
+ return json_parse.loads((await self.__httpx_get_bytes(url=url, params=params, sign_url=sign_url)).decode(encoding="utf-8"))
- async with ClientSession() as session:
- async with session.get(request_url, headers=self.headers, timeout=self.timeout, proxy=self.proxy_container.get()) as request:
- return await request.json()
-
- async def __aiohttp_post_json(self, url: str, params: dict, json: Optional[dict] = None) -> dict:
+ async def __aiohttp_post_json(self, url: str, params: dict, json: Optional[dict] = None, sign_url: bool = False) -> dict:
"""
Post JSON given a URL with parameters
:param url: URL to request data from
:param params: Custom Parameters
:param json: JSON Payload as Dict
+ :param sign_url: Whether to sign the URL (for authenticated endpoints)
:return: JSON Result
-
- :raises: asyncio.TimeoutError
+ :raises: httpx.TimeoutException
"""
- request_url: str = f"{url}?{urllib.parse.urlencode(params if params is not None else dict())}"
- async with ClientSession(cookies=self.cookies) as session:
- async with session.post(request_url, data=json, headers=self.headers, timeout=self.timeout, proxy=self.proxy_container.get()) as request:
- return await request.json()
+ url: str = self.update_url(url, params if params else dict())
+ response: httpx.Response = await self.client.post(await self.__get_signed_url(url) if sign_url else url, data=json, headers=self.headers, timeout=self.timeout)
+ return response.json()
async def get_livestream_page_html(self, unique_id: str) -> str:
"""
@@ -108,26 +129,27 @@ async def get_livestream_page_html(self, unique_id: str) -> str:
:param unique_id: Unique ID of the streamer
:return: HTML string containing page data
- :raises: asyncio.TimeoutError
+ :raises: httpx.TimeoutException
"""
- response: bytes = await self.__aiohttp_get_bytes(f"{TikTokHTTPClient.TIKTOK_URL_WEB}@{unique_id}/live")
+ response: bytes = await self.__httpx_get_bytes(f"{config.TIKTOK_URL_WEB}@{unique_id}/live")
return response.decode(encoding="utf-8")
- async def get_deserialized_object_from_webcast_api(self, path: str, params: dict, schema: str) -> dict:
+ async def get_deserialized_object_from_webcast_api(self, path: str, params: dict, schema: str, sign_url: bool = False) -> dict:
"""
Retrieve and deserialize an object from the Webcast API
+ :param sign_url: Whether to sign the URL (if it's an authenticated request)
:param path: Webcast path
:param params: Parameters to encode into URL
:param schema: Proto schema to decode from
:return: Deserialized data from API in dictionary format
- :raises: asyncio.TimeoutError
+ :raises: httpx.TimeoutException
"""
- response: bytes = await self.__aiohttp_get_bytes(self.TIKTOK_URL_WEBCAST + path, params)
+ response: bytes = await self.__httpx_get_bytes(config.TIKTOK_URL_WEBCAST + path, params, sign_url=sign_url)
return deserialize_message(schema, response)
async def get_json_object_from_webcast_api(self, path: str, params: dict) -> dict:
@@ -137,23 +159,25 @@ async def get_json_object_from_webcast_api(self, path: str, params: dict) -> dic
:param path: Webcast path
:param params: Parameters to encode into URL
:return: JSON data from Webcast API
- :raises: asyncio.TimeoutError
+ :raises: httpx.TimeoutException
"""
- response: dict = await self.__aiohttp_get_json(self.TIKTOK_URL_WEBCAST + path, params)
+ response: dict = await self.__aiohttp_get_json(config.TIKTOK_URL_WEBCAST + path, params)
return response.get("data")
- async def post_json_to_webcast_api(self, path: str, params: dict, json: Optional[dict] = None):
+ async def post_json_to_webcast_api(self, path: str, params: dict, json: Optional[dict] = None, sign_url: bool = False) -> dict:
"""
Post JSON to the Webcast API
+ :param sign_url: Whether to sign the URL (if it's an authenticated request)
:param path: Path to POST
:param params: URLEncoded Params
:param json: JSON Data
- :return: Result
+ :return: Result from the Webcast API POST request
+ :raises: httpx.TimeoutException
"""
- response: dict = await self.__aiohttp_post_json(self.TIKTOK_URL_WEBCAST + path, params, json)
+ response: dict = await self.__aiohttp_post_json(config.TIKTOK_URL_WEBCAST + path, params, json, sign_url=sign_url)
return response
diff --git a/TikTokLive/client/proxy.py b/TikTokLive/client/proxy.py
deleted file mode 100644
index 574fd2d..0000000
--- a/TikTokLive/client/proxy.py
+++ /dev/null
@@ -1,113 +0,0 @@
-import enum
-import random
-from typing import Optional, List
-
-
-class RotationSetting(enum.Enum):
- """
- Rotation settings for a proxy container
-
- """
-
- CONSECUTIVE: int = 1
- """Rotate proxies consecutively, from proxy 0 -> 1 -> 2 -> ...etc."""
-
- RANDOM: int = 2
- """Rotate proxies randomly, from proxy 0 -> 69 -> 420 -> 1 -> ...etc."""
-
- PINNED: int = 3
- """Don't rotate proxies at all, pin to a specific proxy index with set_pinned()"""
-
-
-class ProxyContainer:
-
- def __init__(self, *proxies: str, mode: int = 1, enabled: bool = True):
- """
- Create a ProxyContainer object
-
- :param proxies: *args containing a list of the proxies
- :param mode: The rotation mode as defined in the RotationSetting enum
-
- """
-
- self.proxies: List[str] = list(proxies)
- self.__mode: int = mode
- self.__index: int = 0
- self.__pin: int = 0
- self.__before_pinned: int = self.__mode
- self.__enabled: bool = enabled
-
- @property
- def count(self) -> int:
- """
- Get the current number of proxies in the container
-
- :return: The current number of proxies
-
- """
-
- return len(self.proxies)
-
- def set_enabled(self, enabled: bool) -> None:
- """
- Set whether the system is enabled
-
- :param enabled: Whether to pull a proxy on get()
- :return: None
-
- """
-
- self.__enabled = enabled
-
- def set_pinned(self, index: int) -> None:
- """
- Set the proxy rotator to pinned mode in RotationSetting enum
-
- :param index: Index to pin to
- :return: None
-
- """
-
- self.__pin = index
- self.__before_pinned = self.__mode
- self.__mode = RotationSetting.PINNED
-
- def set_unpinned(self) -> None:
- """
- Remove pinned status and return to whatever mode was set before set_pinned
-
- :return:None
-
- """
-
- self.__mode = self.__before_pinned
-
- def get(self) -> Optional[str]:
- """
- Fetch a proxy using one of the rotation settings defined in RotationSetting
-
- :return: The HTTP/S proxy to return
-
- """
-
- # Has nothing
- if self.count < 1 or not self.__enabled:
- return None
-
- # Consecutive
- if self.__mode == RotationSetting.CONSECUTIVE:
- index: int = self.__index
- if index >= self.count:
- self.__index, index = 1, 0
- else:
- self.__index += 1
-
- # Otherwise random
- else:
- index: int = random.randint(0, self.count - 1)
-
- # Return a proxy
- try:
- return self.proxies[index]
- except IndexError:
- return None
diff --git a/TikTokLive/client/websocket.py b/TikTokLive/client/websocket.py
new file mode 100644
index 0000000..073b948
--- /dev/null
+++ b/TikTokLive/client/websocket.py
@@ -0,0 +1,135 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+import traceback
+from asyncio import AbstractEventLoop
+from typing import TYPE_CHECKING, Optional, Dict, Any
+
+import websockets
+from pyee import AsyncIOEventEmitter
+from websockets.exceptions import ConnectionClosed
+from websockets.legacy.client import WebSocketClientProtocol
+
+from TikTokLive.client.http import TikTokHTTPClient
+from TikTokLive.proto.utilities import deserialize_websocket_message
+
+if TYPE_CHECKING:
+ from TikTokLive.client.base import BaseClient
+
+
+class WebcastWebsocket:
+ """
+ Wrapper class to handle websocket connections to the Webcast API
+
+ """
+
+ def __init__(
+ self,
+ client: BaseClient,
+ ws_url: str,
+ ws_params: Dict[str, str],
+ client_params: Dict[str, str],
+ headers: Dict[str, str],
+ cookies: Dict[str, str],
+ ping_interval_ms: float,
+ loop: AbstractEventLoop,
+ **kwargs
+ ):
+ """
+ Initialize Websocket client for Websocket-based Webcast connections
+
+ :param client: The client to emit events back to as they are received
+ :param ws_url: The URL of the Websocket to connect to
+ :param ws_params: Parameters to be added to the URL that identify the connection to the websocket
+ :param client_params: Various regular parameters to be added to the URL
+ :param headers: Headers to be added to the websocket connection request for authentication
+ :param cookies: Cookies to be added as a "Cookie" header to the websocket connection request for authentication
+ :param loop: The main event loop
+ :param kwargs: Various optional keyword arguments for the websocket itself
+ :param ping_interval_ms: How often to ping the websocket for room data
+
+ """
+
+ # Protected Attributes
+ self._connection: Optional[WebSocketClientProtocol] = None
+ self._websocket_options: Dict[str, Any] = kwargs
+ self._ping_interval: float = ping_interval_ms / 1000
+
+ # Private Attributes
+ self.__cookies: Dict[str, str] = cookies
+ self.__ws_params: Dict[str, str] = {**client_params, **ws_params}
+ self.__headers: Dict[str, str] = {**headers, "Cookie": " ".join(f"{k}={v};" for k, v in cookies.items())}
+ self.__ws_url: str = TikTokHTTPClient.update_url(ws_url, self.__ws_params)
+ self.__loop: AbstractEventLoop = loop
+ self.__client: AsyncIOEventEmitter = client
+
+ async def connect(self) -> bool:
+ """
+ Attempt to connect to the websocket and return the connection status
+
+ :return: Whether the connection was successfully initiated
+ """
+
+ try:
+ # Initiate a connection then keep it open in the connection loop
+ self._connection = await websockets.connect(uri=self.__ws_url, extra_headers=self.__headers, ssl=True, **self._websocket_options)
+ self.__loop.create_task(self.connection_loop())
+ return True
+ except:
+ logging.warning(
+ f"WebcastWebsocket connection failed, will attempt long polling instead. Consider disabling websockets if this persists: "
+ f"\n{traceback.format_exc()}"
+ )
+ return False
+
+ async def connection_loop(self) -> None:
+ """
+ The websocket heartbeat, responsible for making requests to the websocket for room data
+
+ :return: None
+
+ """
+
+ while True:
+ try:
+ # Get a response
+ response: Optional[bytes] = await self._connection.recv()
+
+ # If valid, deserialize response and send back to client
+ if response:
+ self.__client.emit("websocket", deserialize_websocket_message(response))
+
+ except Exception as ex:
+ # If the connection closed, close the websocket
+ if isinstance(ex, ConnectionClosed):
+ await self._connection.close()
+ self.__client.emit("error", ex)
+
+ # Wait until the next ping time
+ await asyncio.sleep(self._ping_interval)
+
+ async def is_open(self) -> bool:
+ """
+ Check whether the websocket connection is open
+
+ :return: The result of the check
+
+ """
+
+ try:
+ await self._connection.ensure_open()
+ except:
+ return False
+
+ return True
+
+ async def close(self) -> None:
+ """
+ Close the websocket connection
+
+ :return: Nothing, you closed it, you dope
+
+ """
+
+ await self._connection.close()
diff --git a/TikTokLive/proto/utilities.py b/TikTokLive/proto/utilities.py
index 0fc6352..f35586e 100644
--- a/TikTokLive/proto/utilities.py
+++ b/TikTokLive/proto/utilities.py
@@ -1,9 +1,8 @@
-from typing import Type, Optional, Any
-
from dacite import from_dict, Config
from dacite.core import T
from dacite.data import Data
from protobuf_to_dict import protobuf_to_dict
+from typing import Type, Optional, Any
from TikTokLive.proto import tiktok_schema_pb2 as tiktok_schema
@@ -55,6 +54,20 @@ def deserialize_message(proto_name: str, obj: bytes) -> dict:
return dict_data
+def deserialize_websocket_message(binary_message: bytes) -> dict:
+ """
+ Deserialize Websocket data. Websocket messages are in a container which contains additional data.
+ A message type 'msg' represents a normal WebcastResponse
+
+ :param binary_message: The binary to decode
+ :return: The resultant decoded python dictionary
+
+ """
+
+ decoded: dict = deserialize_message("WebcastWebsocketMessage", binary_message)
+ return {**decoded, **deserialize_message("WebcastResponse", decoded.get("binary"))} if decoded.get("type") == "msg" else dict()
+
+
def from_dict_plus(data_class: Type[T], data: Data, config: Optional[Config] = None) -> Any:
"""
Load a schema from a dict and set the _as_dict attribute automatically
diff --git a/TikTokLive/types/errors.py b/TikTokLive/types/errors.py
index c1ebd1b..effa1ea 100644
--- a/TikTokLive/types/errors.py
+++ b/TikTokLive/types/errors.py
@@ -30,6 +30,13 @@ class FailedConnection(RuntimeError):
pass
+class InitialCursorMissing(FailedConnection):
+ """
+ Error that is raised when the initial cursor is missing
+
+ """
+ pass
+
class InvalidSessionId(RuntimeError):
"""
Error raised when a session ID is expired or missing
@@ -56,6 +63,13 @@ class ChatMessageRepeat(ChatMessageSendFailure):
pass
+class WebsocketConnectionFailed(RuntimeError):
+ """
+ Raised when a connection to the TikTok Webcast websocket fails
+
+ """
+
+ pass
class FailedHTTPRequest(RuntimeError):
"""
diff --git a/docs/.buildinfo b/docs/.buildinfo
index 848d37a..0426d5c 100644
--- a/docs/.buildinfo
+++ b/docs/.buildinfo
@@ -1,4 +1,4 @@
# Sphinx build info version 1
# This file hashes the configuration used when building these files. When it is not found, a full rebuild will be done.
-config: 3fb376b4bfbd4d683250a5d5dc0817f5
+config: a9fa3a6db386d2ee39d67223e5e6e9ea
tags: 645f666f9bcd5a90fca523b33c5a78b7
diff --git a/docs/README.html b/docs/README.html
index 8513101..c409599 100644
--- a/docs/README.html
+++ b/docs/README.html
@@ -1,118 +1,114 @@
-
+
-
-
+
-
- How To Build — TikTokLive documentation
-
-
-
-
-
-
-
-
+
+
+
+
+
+
-
-
+
+
-
-