From b15387bc8c852b23f4b8ae6374ee5e043b3d9293 Mon Sep 17 00:00:00 2001 From: Gabe Date: Sat, 30 Nov 2024 01:31:14 -0800 Subject: [PATCH] added category search and set stream info --- kick/categories.py | 261 ++++++++++++++++++++++++++++++++++++++- kick/client.py | 60 ++++++++- kick/http.py | 65 ++++++++-- kick/types/categories.py | 53 ++++++++ kick/types/user.py | 8 ++ kick/users.py | 48 ++++++- 6 files changed, 482 insertions(+), 13 deletions(-) diff --git a/kick/categories.py b/kick/categories.py index d9e8022..7544eb4 100644 --- a/kick/categories.py +++ b/kick/categories.py @@ -1,17 +1,174 @@ from __future__ import annotations -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING from .assets import Asset -from .object import HTTPDataclass +from .object import HTTPDataclass, BaseDataclass from .utils import cached_property if TYPE_CHECKING: from .types.categories import Category as CategoryPayload from .types.categories import InnerCategory as ParentCategoryPayload + from .types.categories import ( + Category as CategoryPayload, + InnerCategory as ParentCategoryPayload, + CategoryDocument, + CategorySearchResponse, + TextHighlight as TextHighlightPayload, + SearchHighlight as SearchHighlightPayload, + TextMatchInfo as TextMatchInfoPayload, + CategorySearchHit as CategorySearchHitPayload, + ) + +__all__ = ("Category", "ParentCategory", "SearchCategory", + "CategorySearchResult", "TextHighlight", "SearchHighlight", + "TextMatchInfo", "CategorySearchHit") + + +class TextHighlight(BaseDataclass["TextHighlightPayload"]): + """ + A dataclass representing text highlighting information + Attributes + ----------- + matched_tokens: list[str] + List of tokens that matched + snippet: str + The highlighted text snippet + """ + + @property + def matched_tokens(self) -> list[str]: + return self._data["matched_tokens"] + + @property + def snippet(self) -> str: + return self._data["snippet"] + + def __repr__(self) -> str: + return f"" + + +class SearchHighlight(BaseDataclass["SearchHighlightPayload"]): + """ + A dataclass representing search highlight information + Attributes + ----------- + field: str + The field that was highlighted + matched_tokens: list[str] + List of tokens that matched + snippet: str + The highlighted text snippet + """ + + @property + def field(self) -> str: + return self._data["field"] + + @property + def matched_tokens(self) -> list[str]: + return self._data["matched_tokens"] + + @property + def snippet(self) -> str: + return self._data["snippet"] + + def __repr__(self) -> str: + return f"" + + +class TextMatchInfo(BaseDataclass["TextMatchInfoPayload"]): + """ + A dataclass representing text match scoring information + Attributes + ----------- + best_field_score: str + Score of the best matching field + best_field_weight: int + Weight of the best matching field + fields_matched: int + Number of fields that matched + num_tokens_dropped: int + Number of tokens that were dropped + score: str + Overall match score + tokens_matched: int + Number of tokens that matched + typo_prefix_score: int + Score for typo/prefix matches + """ + + @property + def best_field_score(self) -> str: + return self._data["best_field_score"] + + @property + def best_field_weight(self) -> int: + return self._data["best_field_weight"] + + @property + def fields_matched(self) -> int: + return self._data["fields_matched"] + + @property + def num_tokens_dropped(self) -> int: + return self._data["num_tokens_dropped"] + + @property + def score(self) -> str: + return self._data["score"] + + @property + def tokens_matched(self) -> int: + return self._data["tokens_matched"] + + @property + def typo_prefix_score(self) -> int: + return self._data["typo_prefix_score"] + + def __repr__(self) -> str: + return f"" + + +class CategorySearchHit(BaseDataclass["CategorySearchHitPayload"]): + """ + A dataclass representing a category search hit result + Attributes + ----------- + document: SearchCategory + The matching category document + highlight: dict[str, TextHighlight] + Highlights for each field + highlights: list[SearchHighlight] + List of all highlights + text_match: int + Text match score + text_match_info: TextMatchInfo + Detailed text match information + """ + + @cached_property + def document(self) -> SearchCategory: + return SearchCategory(data=self._data["document"]) + + @cached_property + def highlight(self) -> dict[str, TextHighlight]: + return {k: TextHighlight(data=v) for k, v in self._data["highlight"].items()} -__all__ = ("Category", "ParentCategory") + @cached_property + def highlights(self) -> list[SearchHighlight]: + return [SearchHighlight(data=h) for h in self._data["highlights"]] + + @property + def text_match(self) -> int: + return self._data["text_match"] + @cached_property + def text_match_info(self) -> TextMatchInfo: + return TextMatchInfo(data=self._data["text_match_info"]) + + def __repr__(self) -> str: + return f"" class ParentCategory(HTTPDataclass["ParentCategoryPayload"]): """ @@ -67,6 +224,104 @@ def __eq__(self, other: object) -> bool: def __repr__(self) -> str: return f"" +class SearchCategory(BaseDataclass["CategoryDocument"]): + """ + A dataclass which represents a searchable category on kick + Attributes + ----------- + category_id: int + The id of the parent category. + id: str + The id of the sub-category (game). + name: str + The category's name + slug: str + The category's slug + description: str + The category's description + is_live: bool + Whether the category is live + is_mature: bool + Whether the category is marked as mature + src: str + The category's banner image URL + srcset: str + The category's responsive image srcset + parent: str + The name of the parent category. + """ + + @property + def category_id(self) -> int: + return self._data["category_id"] + + @property + def id(self) -> str: + return self._data["id"] + + @property + def name(self) -> str: + return self._data["name"] + + @property + def slug(self) -> str: + return self._data["slug"] + + @property + def description(self) -> str: + return self._data["description"] + + @property + def is_live(self) -> bool: + return self._data["is_live"] + + @property + def is_mature(self) -> bool: + return self._data["is_mature"] + + @property + def src(self) -> str: + return self._data["src"] + + @property + def srcset(self) -> str: + return self._data["srcset"] + + @property + def parent(self) -> str: + return self._data["parent"] + + def __repr__(self) -> str: + return f"" + + +class CategorySearchResult(BaseDataclass["CategorySearchResponse"]): + """ + A dataclass which represents a category search response + Attributes + ----------- + found: int + Total number of results found + hits: list[SearchCategory] + List of matching categories + page: int + Current page number + """ + + @property + def found(self) -> int: + return self._data["found"] + + @property + def page(self) -> int: + return self._data["page"] + + @cached_property + def hits(self) -> list[CategorySearchHit]: + return [CategorySearchHit(data=hit) for hit in self._data["hits"]] + + def __repr__(self) -> str: + return f"" class Category(HTTPDataclass["CategoryPayload"]): """ diff --git a/kick/client.py b/kick/client.py index 70c24db..fc7d420 100644 --- a/kick/client.py +++ b/kick/client.py @@ -10,7 +10,8 @@ from .http import HTTPClient from .livestream import PartialLivestream from .message import Message -from .users import ClientUser, PartialUser, User, DestinationInfo +from .users import ClientUser, PartialUser, User, DestinationInfo, StreamInfo +from .categories import CategorySearchResult from .utils import MISSING, decorator, setup_logging if TYPE_CHECKING: @@ -235,6 +236,63 @@ async def fetch_stream_url_and_key(self) -> DestinationInfo: data = await self.http.fetch_stream_destination_url_and_key() return DestinationInfo(data=data) + async def set_stream_info( + self, + title: str, + language: str, + subcategory_id: int, + subcategory_name: str | None = None, + is_mature: bool = False, + ) -> StreamInfo: + """ + |coro| + Updates the stream information. + Parameters + ----------- + title: str + The new stream title + language: str + The stream language (e.g. "English") + subcategory_id: int + The ID of the game/category + subcategory_name: Optional[str] + The name of the game/category (optional) + is_mature: bool + Whether the stream is marked as mature content + Raises + ----------- + HTTPException + Failed to update stream information + Returns + ----------- + StreamInfo + The stream info that was set to the logged in user's channel. + """ + + data = await self.http.set_stream_info(title, subcategory_name, subcategory_id, language, is_mature) + return StreamInfo(data=data) + + async def search_categories(self, query: str, /) -> CategorySearchResult: + """ + |coro| + Searches for categories/games on Kick. + Parameters + ----------- + query: str + The search query string + Raises + ----------- + HTTPException + Search request failed + Returns + ----------- + SearchResponse + The search results containing matching categories + """ + + data = await self.http.search_categories(query) + return CategorySearchResult(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 8343e8b..82567b0 100644 --- a/kick/http.py +++ b/kick/http.py @@ -4,6 +4,7 @@ import json import logging from typing import TYPE_CHECKING, Any, Coroutine, Optional, TypeVar, Union +from urllib.parse import urlencode, quote from aiohttp import ClientConnectionError, ClientResponse, ClientSession @@ -20,6 +21,7 @@ from .ws import PusherWebSocket if TYPE_CHECKING: + from types.categories import CategorySearchResponse from types.emotes import EmotesPayload from typing_extensions import Self @@ -48,6 +50,7 @@ ClientUserPayload, UserPayload, DestinationInfoPayload, + StreamInfoPayload ) from .types.videos import GetVideosPayload @@ -95,6 +98,7 @@ async def error_or_nothing(data: Union[dict, str]) -> str: class Route: DOMAIN: str = "https://kick.com" BASE: str = f"{DOMAIN}/api/v2" + SEARCH: str = "https://search.kick.com" def __init__(self, method: str, path: str) -> None: self.path: str = path @@ -109,6 +113,14 @@ def root(cls, method: str, path: str) -> Self: self.url = self.DOMAIN + path return self + @classmethod + def search(cls, method: str, path: str) -> Self: + self = cls.__new__(cls) + self.path = path + self.method = method + self.url = self.SEARCH + path + return self + class HTTPClient: def __init__(self, client: Client): @@ -217,6 +229,23 @@ async def request(self, route: Route, **kwargs) -> Any: headers["Accepts"] = "application/json" cookies = kwargs.pop("cookies", {}) + base_url = kwargs.pop("url", route.url) + + # Handle URL parameters + params = kwargs.get('params') + if params: + encoded_params = urlencode(params, doseq=True) + full_url = f"{base_url}?{encoded_params}" + else: + full_url = base_url + + # Handle bypass URL construction + if not self.whitelisted: + url = f"{self.bypass_host}:{self.bypass_port}/request?url={quote(full_url)}" + # Remove params since they're now in URL + kwargs.pop('params', None) + else: + url = full_url if self.xsrf_token: headers["X-XSRF-TOKEN"] = self.xsrf_token @@ -224,8 +253,6 @@ async def request(self, route: Route, **kwargs) -> Any: if self.token: headers["Authorization"] = f"Bearer {self.token}" - url = route.url - if "json" in kwargs: headers["Content-Type"] = "application/json" @@ -242,11 +269,7 @@ 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, headers=headers, cookies=cookies, **kwargs, @@ -497,6 +520,34 @@ def get_me(self) -> Response[ClientUserPayload]: def fetch_stream_destination_url_and_key(self) -> Response[DestinationInfoPayload]: return self.request(Route.root("GET", "/stream/publish_token")) + def search_categories(self, query: str) -> Response[CategorySearchResponse]: + """Search for categories using the search API""" + route = Route.search("GET", "/collections/subcategory_index/documents/search") + + headers = { + "X-TYPESENSE-API-KEY": "nXIMW0iEN6sMujFYjFuhdrSwVow3pDQu", + } + + params = { + "q": query, + "collections": "subcategory", + "preset": "category_list", + } + + return self.request(route, headers=headers, params=params) + + def set_stream_info(self, title, category_name, category_id, + language, is_mature) -> Response[StreamInfoPayload]: + """Update the stream information""" + route = Route.root("PUT", "/stream/info") + return self.request(route, json={ + "title": title, + "subcategoryName": category_name, + "subcategoryId": category_id, + "language": language, + "is_mature": is_mature + }) + async def get_asset(self, url: str) -> bytes: if self.__session is MISSING: self.__session = ClientSession() diff --git a/kick/types/categories.py b/kick/types/categories.py index d5b1bad..242edca 100644 --- a/kick/types/categories.py +++ b/kick/types/categories.py @@ -1,5 +1,58 @@ +from typing import Any from typing_extensions import TypedDict +class CategoryDocument(TypedDict): + category_id: int + description: str + id: str + is_live: bool + is_mature: bool + name: str + parent: str + slug: str + src: str + srcset: str + + +class TextHighlight(TypedDict): + matched_tokens: list[str] + snippet: str + + +class SearchHighlight(TypedDict): + field: str + matched_tokens: list[str] + snippet: str + + +class TextMatchInfo(TypedDict): + best_field_score: str + best_field_weight: int + fields_matched: int + num_tokens_dropped: int + score: str + tokens_matched: int + typo_prefix_score: int + + +class CategorySearchHit(TypedDict): + document: CategoryDocument + highlight: dict[str, TextHighlight] + highlights: list[SearchHighlight] + text_match: int + text_match_info: TextMatchInfo + + +class CategorySearchResponse(TypedDict): + facet_counts: list[Any] + found: int + hits: list[CategorySearchHit] + out_of: int + page: int + request_params: dict[str, Any] + search_cutoff: bool + search_time_ms: int + class InnerCategory(TypedDict): id: int diff --git a/kick/types/user.py b/kick/types/user.py index bd7a1d9..0fc196d 100644 --- a/kick/types/user.py +++ b/kick/types/user.py @@ -107,6 +107,14 @@ class ClientUserStreamerChannelsPayload(TypedDict): verified: None # Unknown +class StreamInfoPayload(TypedDict): + title: str + subcategoryId: int + subcategoryName: None | str + language: str + is_mature: bool + + class ClientUserPayload(TypedDict): id: int email: str diff --git a/kick/users.py b/kick/users.py index 6fa4545..ab1b3d2 100644 --- a/kick/users.py +++ b/kick/users.py @@ -17,9 +17,10 @@ from .chatroom import Chatroom from .http import HTTPClient from .types.user import (ClientUserPayload, InnerUser, UserPayload, - DestinationInfoPayload) + DestinationInfoPayload, StreamInfoPayload) -__all__ = ("DestinationInfo", "Socials", "PartialUser", "User", "ClientUser") +__all__ = ("DestinationInfo", "Socials", "PartialUser", "User", "ClientUser", + "StreamInfo") class DestinationInfo(BaseDataclass["DestinationInfoPayload"]): @@ -384,6 +385,49 @@ async def stop_watching(self) -> None: self.http.client._watched_users.pop(self.channel_id) + class StreamInfo(BaseDataclass["StreamInfoPayload"]): + """ + A dataclass which represents stream information + Attributes + ----------- + title: str + The stream title + subcategory_id: int + The ID of the game/category + subcategory_name: Optional[str] + The name of the game/category + language: str + The stream language + is_mature: bool + Whether the stream is marked as mature content + """ + + @property + def title(self) -> str: + return self._data["title"] + + @property + def subcategory_id(self) -> int: + return self._data["subcategoryId"] + + @property + def subcategory_name(self) -> str | None: + return self._data.get("subcategoryName") + + @property + def language(self) -> str: + return self._data["language"] + + @property + def is_mature(self) -> bool: + return self._data["is_mature"] + + def __repr__(self) -> str: + return f"" + + + + class ClientUser(BaseUser): def __init__(self, *, data: ClientUserPayload, http: HTTPClient) -> None: self._data = data