diff --git a/README.md b/README.md index 7ddaf46..90bfafd 100644 --- a/README.md +++ b/README.md @@ -27,12 +27,17 @@ audio_stream = video.get_streams('audio')[0] print(f"Audio stream URL: {audio_stream.url} ({audio_stream.mime_type})") ``` +You can find more examples in the [`tests`](https://github.com/CWKevo/python-piped-api-client/tree/master/tests) folder. + ## Why? + -The creation of this package was also primarily fueled by the same type of motivation [Piped has](https://piped-docs.kavin.rocks/docs/why/). +The creation of this package was primarily fueled by the same type of motivation [Piped has](https://piped-docs.kavin.rocks/docs/why/). Google's API is not very easy-to-use - you must obtain some JSON thingy to use it, and it is very low-level and not very user-friendly. On the other hand, this package accessed the [Piped API](https://piped.kavin.rocks/), which has a much more high-level API and doesn't need an account or API keys. diff --git a/piped_api/client.py b/piped_api/client.py index 6ad9661..3b165b1 100644 --- a/piped_api/client.py +++ b/piped_api/client.py @@ -5,7 +5,7 @@ from requests import Session from .models import BasePipedModel from .models.comments import Comments from .models.videos import Video -from .models.channels import Channel +from .models.channels import NextPageChannel, Channel _MDL = t.TypeVar('_MDL', bound=t.Type[BasePipedModel]) @@ -35,7 +35,7 @@ class PipedClient: - def _get_json(self, uri: str, as_model: t.Optional[_MDL]=None, **kwargs) -> t.Union[_MDL, t.Dict[str, t.Any]]: + def _get_json(self, uri: str, as_model: t.Optional[_MDL]=None, **kwargs) -> t.Union[_MDL, t.Dict[str, t.Any], t.List[t.Any]]: """ Obtains JSON data from specific URI of the Piped API. @@ -56,6 +56,19 @@ class PipedClient: return json + def get_video(self, video_id: str, **kwargs) -> Video: + """ + Gets information about a specific video. + + ### Parameters: + - `video_id` - The ID of the video to get information for + - `**kwargs` - Additional keyword arguments to pass to `requests.Session.get` + + [Piped Documentation](https://piped-docs.kavin.rocks/docs/api-documentation/#streamsvideoid) + """ + + return self._get_json(f"/streams/{video_id}", Video, **kwargs) + def get_comments(self, video_id: str, nextpage: t.Optional[t.Dict[str, t.Optional[str]]]=None, **kwargs) -> Comments: """ @@ -66,29 +79,15 @@ class PipedClient: - `nextpage` - Nextpage data, obtained from `.models.comments.Comments.nextpage` property. If this is `None`, the first page of comments is returned. There are often 20 comments per page. - `**kwargs` - Additional keyword arguments to pass to `requests.Session.get` - """ - kw = kwargs.copy() + [Piped Documentation](https://piped-docs.kavin.rocks/docs/api-documentation/#commentsvideoid) + """ if nextpage is not None: - kw.update({'params': {'nextpage': nextpage}}) - return self._get_json(f"/nextpage/comments/{video_id}", Comments, **kw) + kwargs.update({'params': {'nextpage': nextpage}}) + return self._get_json(f"/nextpage/comments/{video_id}", Comments, **kwargs) - else: - return self._get_json(f"/comments/{video_id}", Comments, **kw) - - - - def get_video(self, video_id: str, **kwargs) -> Video: - """ - Gets information about a specific video. - - ### Parameters: - - `video_id` - The ID of the video to get information for - - `**kwargs` - Additional keyword arguments to pass to `requests.Session.get` - """ - - return self._get_json(f"/streams/{video_id}", Video, **kwargs) + return self._get_json(f"/comments/{video_id}", Comments, **kwargs) def get_trending(self, country_code: str='US', **kwargs) -> t.List[Video.RelatedStream]: @@ -100,23 +99,30 @@ class PipedClient: - `country_code` - The country code ([ISO 3166-1 alpha-2](https://en.wikipedia.org/wiki/ISO_3166-1_alpha-2#Officially_assigned_code_elements)) to get trending videos for. This is automatically capitalized by this package, since Piped for some reason doesn't accept lowercase country codes. Note: countries such as China or North Korea don't have trending videos, so they will always return an empty list. - `**kwargs` - Additional keyword arguments to pass to `requests.Session.get` + + [Piped Documentation](https://piped-docs.kavin.rocks/docs/api-documentation/#trending) """ - kw = kwargs.copy() - kw.update({'params': {'region': country_code.upper()}}) + kwargs.update({'params': {'region': country_code.upper()}}) - return [Video.RelatedStream(trending_video) for trending_video in self._get_json(f"/trending", **kw)] + return [Video.RelatedStream(trending_video) for trending_video in self._get_json(f"/trending", **kwargs)] - def get_channel_by_id(self, channel_id: str, **kwargs) -> Channel: + def get_channel_by_id(self, channel_id: str, nextpage: t.Optional[t.Dict[str, t.Optional[str]]]=None, **kwargs) -> t.Union[NextPageChannel, Channel]: """ Gets information about a specific channel by its ID. ### Parameters: - `channel_id` - The ID of the channel to get information for - `**kwargs` - Additional keyword arguments to pass to `requests.Session.get` + + [Piped Documentation](https://piped-docs.kavin.rocks/docs/api-documentation/#channelchannelid) """ + if nextpage is not None: + kwargs.update({'params': {'nextpage': nextpage}}) + return self._get_json(f"/nextpage/channel/{channel_id}", NextPageChannel, **kwargs) + return self._get_json(f"/channel/{channel_id}", Channel, **kwargs) @@ -128,6 +134,24 @@ class PipedClient: ### Parameters: - `channel_name` - The name of the channel to get information for - `**kwargs` - Additional keyword arguments to pass to `requests.Session.get` + + [Piped Documentation](https://piped-docs.kavin.rocks/docs/api-documentation/#cname) """ return self._get_json(f"/c/{channel_name}", Channel, **kwargs) + + + def get_search_suggestions(self, search_query: str, **kwargs) -> t.List[str]: + """ + Obtains search suggestions for a query. + + ### Parameters: + - `search_query` - The query to get search suggestions for + - `**kwargs` - Additional keyword arguments to pass to `requests.Session.get` + + [Piped Documentation](https://piped-docs.kavin.rocks/docs/api-documentation/#suggestions) + """ + + kwargs.update({'params': {'query': search_query}}) + + return self._get_json(f"/suggestions", **kwargs) diff --git a/piped_api/models/channels.py b/piped_api/models/channels.py index a5e5801..9f38148 100644 --- a/piped_api/models/channels.py +++ b/piped_api/models/channels.py @@ -5,9 +5,40 @@ from .videos import Video -class Channel(BasePipedModel): +class NextPageChannel(BasePipedModel): """ - Represents a YouTube channel + Represents a channel obtained via the `nextpage` endpoint. + + This model contains only `nextpage` and `relatedStreams`. It's a parent for `Channel`. + """ + + @property + def nextpage(self) -> str: + """ + A JSON encoded string to be passed to the `'nextpage'` endpoint(s) when + obtaining paginated data. + """ + + return self.data['nextpage'] + + + @property + def uploaded_videos(self) -> t.List[Video.RelatedStream]: + """ + List of uploaded videos from the current fetched data + + There are max. 30 videos per page + """ + + return [Video.RelatedStream(video_data) for video_data in self.data['relatedStreams']] + + + +class Channel(NextPageChannel): + """ + Represents a YouTube channel. + + Contains properties of `NextPageChannel`. """ @property @@ -55,16 +86,6 @@ class Channel(BasePipedModel): return self.data['description'] - @property - def nextpage(self) -> str: - """ - A JSON encoded string to be passed to the `'nextpage'` endpoint(s) when - obtaining paginated data. - """ - - return self.data['nextpage'] - - @property def subscriber_count(self) -> int: """ @@ -81,12 +102,3 @@ class Channel(BasePipedModel): """ return self.data['verified'] - - - @property - def uploaded_videos(self) -> t.List[Video.RelatedStream]: - """ - List of uploaded videos from the current fetched data - """ - - return [Video.RelatedStream(video_data) for video_data in self.data['relatedVideos']] diff --git a/piped_api/models/videos.py b/piped_api/models/videos.py index 726690d..ce8e8ad 100644 --- a/piped_api/models/videos.py +++ b/piped_api/models/videos.py @@ -432,7 +432,7 @@ class Video(BasePipedModel): List of related streams """ - return [self.RelatedStream(video_data) for video_data in self.data['relatedVideos']] + return [self.RelatedStream(video_data) for video_data in self.data['relatedStreams']] diff --git a/tests/test_channel.py b/tests/test_channel.py deleted file mode 100644 index da18eb1..0000000 --- a/tests/test_channel.py +++ /dev/null @@ -1,39 +0,0 @@ -from tests import CLIENT - - -def test_channel_by_id(channel_id: str='UCuAXFkgsw1L7xaCfnd5JJOw') -> None: - """ - Prints out information about a channel by its ID. - """ - - channel = CLIENT.get_channel_by_id(channel_id) - assert channel.id == channel_id - - print(f""" - Channel ID: {channel_id} - Name: {channel.name} - Description: {channel.description} - Subscriber count: {channel.subscriber_count} - """) - - - -def test_channel_by_name(channel_name: str='SusanWojcicki') -> None: - """ - Prints out information about a channel by its ID. - """ - - channel = CLIENT.get_channel_by_name(channel_name) - - print(f""" - Channel ID: {channel.id} - Name: {channel.name} - Description: {channel.description} - Subscriber count: {channel.subscriber_count} - """) - - - -if __name__ == '__main__': - test_channel_by_id() - test_channel_by_name() diff --git a/tests/test_channels.py b/tests/test_channels.py new file mode 100644 index 0000000..c0ddaad --- /dev/null +++ b/tests/test_channels.py @@ -0,0 +1,58 @@ +from tests import CLIENT + +from datetime import timedelta +from random import choice + + +def test_channel_by_id(channel_id: str='UCuAXFkgsw1L7xaCfnd5JJOw') -> None: + """ + Prints out information about a channel by its ID. + """ + + channel = CLIENT.get_channel_by_id(channel_id) + + video_id = channel.uploaded_videos[0].url.removeprefix('/watch?v=') + likes = CLIENT.get_video(video_id).likes + + print(f"Total likes for last video of {channel.name}: {likes}") + + + +def test_channel_by_name(channel_name: str='SusanWojcicki') -> None: + """ + Prints out information about a channel by its ID. + """ + + channel = CLIENT.get_channel_by_name(channel_name) + + print(f""" + Channel ID: {channel.id} + Name: {channel.name} + Description: {channel.description} + Subscriber count: {channel.subscriber_count} + """) + + + +def test_get_watchtime_trending() -> None: + """ + Prints out the total watchtime for recent videos of a random trending channel + """ + + trending_channel_id = choice(CLIENT.get_trending('SK')).uploader_url.removeprefix('/channel/') + trending_channel = CLIENT.get_channel_by_id(trending_channel_id) + + + total_watchtime = timedelta(milliseconds=0) + + for video in trending_channel.uploaded_videos: + total_watchtime += video.duration + + print(f"Total watchtime for recent {len(trending_channel.uploaded_videos)} videos of {trending_channel.name} (https://youtube.com/channel/{trending_channel.id}): {total_watchtime}") + + + +if __name__ == '__main__': + test_channel_by_name() + test_channel_by_id() + test_get_watchtime_trending() diff --git a/tests/test_suggestions.py b/tests/test_suggestions.py new file mode 100644 index 0000000..1eaaa95 --- /dev/null +++ b/tests/test_suggestions.py @@ -0,0 +1,17 @@ +from tests import CLIENT + + +def test_suggestions(search_query: str='Susan') -> None: + """ + Obtains search suggestions for a query. + """ + + suggestions = CLIENT.get_search_suggestions(search_query) + + assert len(suggestions) > 0 + print(suggestions) + + + +if __name__ == '__main__': + test_suggestions()