cere
/
python-piped-api-client
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Some tweaks

This commit is contained in:
Kevo 2022-02-28 09:24:59 +01:00
parent d4ab82c07e
commit 8f8ffe3e7b
7 changed files with 164 additions and 87 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
README.md
View File

@ -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?
<!-- Soon... maybe.
This package has allowed me to start creating my open-source project, [ArchiveTube](https://github.com/CWKevo/ArchiveTube) - a scrapper and archive for YouTube content (videos and comments) - to preserve them and make them available to anyone, with ability to search for comments and videos. View hall of fame (most liked comments and videos), bring back dislikes via [ReturnYouTubeDislike.com](https://returnyoutubedislike.com), view deleted content and much more!
Google has showed us that they make YouTube own us by harvesting our data. This is also followed by non-throught out decisions, which their users aren't happy with. Let's do it the other way around this time by reclaiming our content and entertainment back & make YouTube great again!
-->
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.

74
piped_api/client.py
View File

@ -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)

54
piped_api/models/channels.py
View File

@ -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']]

2
piped_api/models/videos.py
View File

@ -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']]

39
tests/test_channel.py
View File

@ -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()

58
tests/test_channels.py Normal file
View File

@ -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()

17
tests/test_suggestions.py Normal file
View File

@ -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()