diff --git a/kick/client.py b/kick/client.py index 874fd75..70c24db 100644 --- a/kick/client.py +++ b/kick/client.py @@ -10,7 +10,7 @@ from .http import HTTPClient from .livestream import PartialLivestream from .message import Message -from .users import ClientUser, PartialUser, User +from .users import ClientUser, PartialUser, User, DestinationInfo from .utils import MISSING, decorator, setup_logging if TYPE_CHECKING: @@ -213,6 +213,28 @@ async def fetch_user(self, name: str, /) -> User: user = User(data=data, http=self.http) return user + async def fetch_stream_url_and_key(self) -> DestinationInfo: + """ + |coro| + + Fetches your stream URL and stream key from the API. + You must be authenticated to use this endpoint. + + Raises + ----------- + HTTPException + Fetching Failed + Forbidden + You are not authenticated + + Returns + ----------- + DestinationInfo + """ + + data = await self.http.fetch_stream_destination_url_and_key() + return DestinationInfo(data=data) + def dispatch(self, event_name: str, *args, **kwargs) -> None: event_name = f"on_{event_name}" diff --git a/kick/http.py b/kick/http.py index 57df00c..8343e8b 100644 --- a/kick/http.py +++ b/kick/http.py @@ -43,7 +43,12 @@ ReplyOriginalSender, V1MessageSentPayload, ) - from .types.user import ChatterPayload, ClientUserPayload, UserPayload + from .types.user import ( + ChatterPayload, + ClientUserPayload, + UserPayload, + DestinationInfoPayload, + ) from .types.videos import GetVideosPayload T = TypeVar("T") @@ -60,10 +65,9 @@ async def json_or_text(response: ClientResponse, /) -> Union[dict[str, Any], str]: text = await response.text() try: - try: - return json.loads(text) - except json.JSONDecodeError: - pass + return json.loads(text) + except json.JSONDecodeError: + pass except KeyError: pass @@ -238,9 +242,11 @@ async def request(self, route: Route, **kwargs) -> Any: try: res = await self.__session.request( route.method, - url - if self.whitelisted is True - else f"{self.bypass_host}:{self.bypass_port}/request?url={url}", + ( + url + if self.whitelisted is True + else f"{self.bypass_host}:{self.bypass_port}/request?url={url}" + ), headers=headers, cookies=cookies, **kwargs, @@ -488,6 +494,9 @@ def reply_to_message( def get_me(self) -> Response[ClientUserPayload]: return self.request(Route.root("GET", "/api/v1/user")) + def fetch_stream_destination_url_and_key(self) -> Response[DestinationInfoPayload]: + return self.request(Route.root("GET", "/stream/publish_token")) + async def get_asset(self, url: str) -> bytes: if self.__session is MISSING: self.__session = ClientSession() diff --git a/kick/types/user.py b/kick/types/user.py index c292199..bd7a1d9 100644 --- a/kick/types/user.py +++ b/kick/types/user.py @@ -140,3 +140,8 @@ class ClientUserPayload(TypedDict): streamer_channel: ClientUserStreamerChannelsPayload roles: list # Unknown profilepic: str | None + + +class DestinationInfoPayload(TypedDict): + rtmp_publish_path: str + rtmp_stream_token: str diff --git a/kick/users.py b/kick/users.py index 9011c0f..6fa4545 100644 --- a/kick/users.py +++ b/kick/users.py @@ -16,9 +16,33 @@ if TYPE_CHECKING: from .chatroom import Chatroom from .http import HTTPClient - from .types.user import ClientUserPayload, InnerUser, UserPayload + from .types.user import (ClientUserPayload, InnerUser, UserPayload, + DestinationInfoPayload) -__all__ = ("User", "Socials", "PartialUser", "ClientUser") +__all__ = ("DestinationInfo", "Socials", "PartialUser", "User", "ClientUser") + + +class DestinationInfo(BaseDataclass["DestinationInfoPayload"]): + """ + Information about a user's stream destination + + Attributes + ----------- + stream_url: str + The URL for streaming + stream_key: str + The stream key + """ + + @property + def stream_url(self) -> str: + """The URL for streaming""" + return self._data["rtmp_publish_path"] + + @property + def stream_key(self) -> str: + """The stream key""" + return self._data["rtmp_stream_token"] class Socials(BaseDataclass["InnerUser | ClientUserPayload"]):