From 545fe5d98c1fbc25d88cb5283b94023316930982 Mon Sep 17 00:00:00 2001 From: chillymosh <86857777+chillymosh@users.noreply.github.com> Date: Wed, 20 Mar 2024 21:27:03 +0000 Subject: [PATCH] Add Video methods models --- twitchio/client.py | 106 ++++++++++++++++++++++++++++++++++- twitchio/http.py | 40 ++++++++++++- twitchio/models.py | 87 ++++++++++++++++++++++++++++ twitchio/types_/responses.py | 45 ++++++++++++++- 4 files changed, 273 insertions(+), 5 deletions(-) diff --git a/twitchio/client.py b/twitchio/client.py index 9991f122..6ede5f16 100644 --- a/twitchio/client.py +++ b/twitchio/client.py @@ -43,6 +43,7 @@ SearchChannel, Stream, Team, + Video, ) from .payloads import EventErrorPayload from .web import AiohttpAdapter @@ -132,7 +133,8 @@ def dispatch(self, event: str, payload: Any | None = None) -> None: _ = [asyncio.create_task(self._dispatch(listener, original=payload)) for listener in listeners] - async def setup_hook(self) -> None: ... + async def setup_hook(self) -> None: + ... async def login(self, *, token: str | None = None) -> None: if not token: @@ -335,8 +337,11 @@ async def fetch_clips( :class:`~twitchio.HTTPAsyncIterator`[:class:`~twitchio.Clip`] """ - if sum(x is not None for x in [broadcaster_id, game_id, clip_ids]) > 1: - raise ValueError("The parameters 'broadcaster_id', 'game_id', and 'ids' are mutually exclusive.") + provided: int = len([v for v in (broadcaster_id, game_id, clip_ids) if v]) + if provided > 1: + raise ValueError("Only one of 'name', 'id', or 'igdb_id' can be provided.") + elif provided == 0: + raise ValueError("One of 'name', 'id', or 'igdb_id' must be provided.") first = max(1, min(100, first)) @@ -617,6 +622,101 @@ async def search_channels( token_for=token_for, ) + async def fetch_videos( + self, + *, + ids: list[str | int] | None = None, + user_id: str | int | None = None, + game_id: str | int | None = None, + language: str | None = None, + period: Literal["all", "day", "month", "week"] = "all", + sort: Literal["time", "trending", "views"] = "time", + type: Literal["all", "archive", "highlight", "upload"] = "all", + first: int = 20, + token_for: str | None = None, + ) -> HTTPAsyncIterator[Video]: + """Fetch a list of [`twitchio.Video`][twitchio.Video] objects with the provided `ids`, `user_id` or `game_id`. + + One of `ids`, `user_id` or `game_id` must be provided. + If more than one is provided or no parameters are provided, a `ValueError` will be raised. + + + Parameters + ---------- + ids: list[str | int] | None + A list of video IDs to fetch. + user_id: str | int | None + The ID of the user whose list of videos you want to get. + game_id: str | int | None + The igdb_id of the game to fetch. + language: str | None + period: Literal["all", "day", "month", "week"] + sort: Literal["time", "trending", "views"] + type: Literal["all", "archive", "highlight", "upload"] + first: int + token_for: str | None + An optional User OAuth token to use instead of the default app token. + + Returns + ------- + list[Video] + A list of Video objects if found. + + Raises + ------ + ValueError + Only one of the 'ids', 'user_id', or 'game_id' parameters can be provided. + ValueError + One of the 'ids', 'user_id', or 'game_id' parameters must be provided. + """ + provided: int = len([v for v in (ids, game_id, user_id) if v]) + if provided > 1: + raise ValueError("Only one of 'ids', 'user_id', or 'game_id' can be provided.") + elif provided == 0: + raise ValueError("One of 'name', 'id', or 'igdb_id' must be provided.") + + first = max(1, min(100, first)) + + return await self._http.get_videos( + ids=ids, + user_id=user_id, + game_id=game_id, + language=language, + period=period, + sort=sort, + type=type, + first=first, + token_for=token_for, + ) + + async def delete_videos(self, ids: list[str | int], token_for: str) -> list[str]: + """Deletes one or more videos. You may delete past broadcasts, highlights, or uploads. + + This requires a user token with the scope ``channel:manage:videos``. + The limit is to delete 5 ids at a time, so if more than 5 ids are provided we will attempt to delete them in chunks. + If any of the videos fail to delete in the request then none will be deleted in that chunk. + + Parameters + ---------- + ids: list[str | int] | None + A list of video IDs to fetch. + token_for: str + A User OAuth token with the scope ``channel:manage:videos``. + + Returns + ------- + list[str] + A list of Video IDs that were successfully deleted. + """ + resp: list[str] = [] + + for chunk in [ids[x : x + 5] for x in range(0, len(ids), 5)]: + data = await self._http.delete_videos(ids=chunk, token_for=token_for) + if data: + resp.extend(data["data"]) + + return resp + def doc_test(self, thing: int = 1) -> int: """This is a test method to test and view certain elements of the mkdocs style documentation. diff --git a/twitchio/http.py b/twitchio/http.py index 09f8828a..be9a8609 100644 --- a/twitchio/http.py +++ b/twitchio/http.py @@ -36,7 +36,7 @@ from . import __version__ from .exceptions import HTTPException -from .models import Clip, Game, SearchChannel, Stream +from .models import Clip, Game, SearchChannel, Stream, Video from .utils import _from_json # type: ignore @@ -57,6 +57,8 @@ SearchChannelResponse, StreamResponse, TeamPayload, + VideoDeletePayload, + VideoResponse, ) @@ -557,3 +559,39 @@ async def converter(data: GameResponse) -> Game: iterator: HTTPAsyncIterator[Game] = self.request_paginated(route, converter=converter) return iterator + + async def get_videos( + self, + ids: list[str | int] | None = None, + user_id: str | int | None = None, + game_id: str | int | None = None, + language: str | None = None, + period: Literal["all", "day", "month", "week"] = "all", + sort: Literal["time", "trending", "views"] = "time", + type: Literal["all", "archive", "highlight", "upload"] = "all", + first: int = 20, + token_for: str | None = None, + ) -> HTTPAsyncIterator[Video]: + params: dict[str, int | str | list[str | int]] = {"first": first, "period": period, "sort": sort, "type": type} + + if ids is not None: + params["id"] = ids + if user_id is not None: + params["user_id"] = user_id + if game_id is not None: + params["game_id"] = game_id + if language is not None: + params["language"] = language + + route = Route("GET", "videos", params=params, token_for=token_for) + + async def converter(data: VideoResponse) -> Video: + return Video(data, http=self) + + iterator = self.request_paginated(route, converter=converter) + return iterator + + async def delete_videos(self, ids: list[str | int], token_for: str) -> VideoDeletePayload: + params = {"id": ids} + route: Route = Route("DELETE", "videos", params=params, token_for=token_for) + return await self.request_json(route) diff --git a/twitchio/models.py b/twitchio/models.py index eab943c7..a679e9e5 100644 --- a/twitchio/models.py +++ b/twitchio/models.py @@ -48,6 +48,7 @@ SearchChannelResponse, StreamResponse, TeamResponse, + VideoResponse, ) @@ -63,6 +64,7 @@ "SearchChannel", "Stream", "Team", + "Video", ) @@ -657,3 +659,88 @@ def __eq__(self, __value: object) -> bool: return NotImplemented return __value.id == self.id + + +class Video: + """ + Represents video information + + Attributes + ----------- + id: :class:`int` + The ID of the video. + user: :class:`~twitchio.PartialUser` + User who owns the video. + title: :class:`str` + Title of the video + description: :class:`str` + Description of the video. + created_at: :class:`datetime.datetime` + Date when the video was created. + published_at: :class:`datetime.datetime` + Date when the video was published. + url: :class:`str` + URL of the video. + thumbnail_url: :class:`str` + Template URL for the thumbnail of the video. + viewable: :class:`str` + Indicates whether the video is public or private. + view_count: :class:`int` + Number of times the video has been viewed. + language: :class:`str` + Language of the video. + type: :class:`str` + The type of video. + duration: :class:`str` + Length of the video. + """ + + __slots__ = ( + "_http", + "id", + "user", + "title", + "description", + "created_at", + "published_at", + "url", + "thumbnail_url", + "viewable", + "view_count", + "language", + "type", + "duration", + ) + + def __init__(self, data: VideoResponse, *, http: HTTPClient) -> None: + self._http: HTTPClient = http + self.id: str = data["id"] + self.user = PartialUser(data["user_id"], data["user_name"]) + self.title: str = data["title"] + self.description: str = data["description"] + self.created_at = parse_timestamp(data["created_at"]) + self.published_at = parse_timestamp(data["published_at"]) + self.url: str = data["url"] + self.thumbnail_url: str = data["thumbnail_url"] + self.viewable: str = data["viewable"] + self.view_count: int = data["view_count"] + self.language: str = data["language"] + self.type: str = data["type"] + self.duration: str = data["duration"] + + def __repr__(self) -> str: + return f"