added category search and set stream info

Author: gub-7

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"<TextHighlight matched_tokens={self.matched_tokens!r} snippet={self.snippet!r}>"
+
+
+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"<SearchHighlight field={self.field!r} matched_tokens={self.matched_tokens!r} snippet={self.snippet!r}>"
+
+
+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"<TextMatchInfo score={self.score!r} tokens_matched={self.tokens_matched} fields_matched={self.fields_matched}>"
+
+
+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"<CategorySearchHit document={self.document!r} text_match={self.text_match}>"
 
 class ParentCategory(HTTPDataclass["ParentCategoryPayload"]):
     """
@@ -67,6 +224,104 @@ def __eq__(self, other: object) -> bool:
     def __repr__(self) -> str:
         return f"<ParentCategory id={self.id!r} name={self.name!r} icon={self.icon!r}>"
 
+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"<SearchCategory name={self.name!r} slug={self.slug!r} is_live={self.is_live}>"
+
+
+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"<CategorySearchResult found={self.found} page={self.page} hits={len(self.hits)}>"
 
 class Category(HTTPDataclass["CategoryPayload"]):
     """

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}"