cere/plugin.video.piped
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Initial Commit

Browse source
This commit is contained in:
cere 2024-01-16 05:08:13 -05:00
commit 0543b89d8f
12 changed files with 1256 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

25
lib/piped_api/__init__.py Normal file
View file

@ -0,0 +1,25 @@
import typing as t
from pathlib import Path
from .client import PipedClient
from .models.comments import Comments
# For pdoc:
README_PATH = Path(__file__).parent.parent.absolute() / Path('README.md')
try:
with open(README_PATH, 'r', encoding="UTF-8") as readme:
__readme__ = readme.read()
except:
__readme__ = "Failed to read README.md!"
__doc__ = __readme__
# Supress unused-import warnings:
if t.TYPE_CHECKING:
_ = [PipedClient, Comments]

170
lib/piped_api/client.py Normal file
View file

@ -0,0 +1,170 @@
import typing as t
from requests import Session
from .models import BasePipedModel
from .models.comments import Comments
from .models.videos import Video
from .models.channels import NextPageChannel, Channel
_MDL = t.TypeVar('_MDL', bound=t.Type[BasePipedModel])
class APIError(Exception): """Raised when an API call fails"""
class PipedClient:
"""
An API client for [Piped](https://piped.kavin.rocks).
See also [Piped API docs](https://piped-docs.kavin.rocks/docs)
"""
def __init__(self, base_api_url: str='https://pipedapi.kavin.rocks', session: t.Type[Session]=Session()) -> None:
"""
### Parameters:
- `base_api_url` - The base URL to the instance's API. Trailing slashes will be stripped.
- `session` - A class/subclass of `requests.Session` to use for all requests.
For example, you could use [requests-cache](https://pypi.org/project/requests-cache/) to make all requests cacheable.
"""
self.base_api_url = base_api_url.strip("/")
self.session = session
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.
### Parameters:
- `uri` - The URI to get JSON data from
- `as_model` - The `BasePipedModel` to load the JSON data into. If this is `None`, the JSON data is returned as a `dict`.
- `**kwargs` - Additional keyword arguments to pass to `requests.Session.get`
"""
json: t.Union[dict, list] = self.session.get(f"{self.base_api_url}{uri}", **kwargs).json()
if isinstance(json, dict) and json.get('error', None) is not None:
raise APIError(f"Error: {json['error']}")
if as_model is not None:
return as_model(json)
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:
"""
Gets a list of comments for a specific video.
### Parameters:
- `video_id` - The ID of the video to get comments for
- `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`
[Piped Documentation](https://piped-docs.kavin.rocks/docs/api-documentation/#commentsvideoid)
"""
if nextpage is not None:
kwargs.update({'params': {'nextpage': nextpage}})
return self._get_json(f"/nextpage/comments/{video_id}", Comments, **kwargs)
return self._get_json(f"/comments/{video_id}", Comments, **kwargs)
def get_trending(self, country_code: str='US', **kwargs) -> t.List[Video.RelatedStream]:
"""
Obtains trending videos for a specific country. If there are no trending videos (or `country_code` is invalid),
an empty list is returned.
### Parameters:
- `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)
"""
kwargs.update({'params': {'region': country_code.upper()}})
return [Video.RelatedStream(trending_video) for trending_video in self._get_json(f"/trending", **kwargs)]
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)
def get_channel_by_name(self, channel_name: str, **kwargs) -> Channel:
"""
Gets information about a specific channel by its name.
### 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)
def get_search_results(self, search_query: str, filter: str="all", **kwargs) -> t.List[str]:
"""
Obtains search results for a query.
### Parameters:
- `search_query` - The query to get search results for
- `**kwargs` - Additional keyword arguments to pass to `requests.Session.get`
"""
kwargs.update({'params': {'q': search_query, 'filter': filter}})
return self._get_json(f"/search", **kwargs)

15
lib/piped_api/models/__init__.py Normal file
View file

@ -0,0 +1,15 @@
import typing as t
class BasePipedModel:
"""
Base class for all Piped models.
"""
def __init__(self, data: t.Dict[str, t.Any]) -> None:
"""
### Parameters:
- `data` - The JSON (`dict`) data to initialize the model with.
"""
self.data = data

104
lib/piped_api/models/channels.py Normal file
View file

@ -0,0 +1,104 @@
import typing as t
from . import BasePipedModel
from .videos import Video
class NextPageChannel(BasePipedModel):
"""
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
def id(self) -> str:
"""
The channel's ID
"""
return self.data['id']
@property
def name(self) -> str:
"""
The channel's name
"""
return self.data['name']
@property
def avatar_url(self) -> str:
"""
The channel's avatar URL
"""
return self.data['avatarUrl']
@property
def banner_url(self) -> str:
"""
The channel's banner URL
"""
return self.data['bannerUrl']
@property
def description(self) -> str:
"""
The channel's description
"""
return self.data['description']
@property
def subscriber_count(self) -> int:
"""
The number of subscribers the channel has
"""
return self.data['subscriberCount']
@property
def verified(self) -> bool:
"""
Whether or not the channel is verified by YouTube (has a badge)
"""
return self.data['verified']

163
lib/piped_api/models/comments.py Normal file
View file

@ -0,0 +1,163 @@
import typing as t
from . import BasePipedModel
class Comments(BasePipedModel):
class Comment(BasePipedModel):
@property
def author(self) -> str:
"""
The name of the author of the comment
"""
return self.data['author']
@property
def comment_id(self) -> str:
"""
The comment ID
"""
return self.data['commentId']
@property
def comment_text(self) -> str:
"""
The text of the comment
"""
return self.data['commentText']
@property
def commented_time(self) -> str:
"""
The time the comment was made (format: `'x y ago'`).
### Note:
The raw time from API also includes the `'(edited)'` suffix to mark comment as edited (if it was).
By accessing this property, the suffix is automatically removed.
If you for whatever reason want to keep the suffix, access this property directly via `Comment.data['commentedTime']`
"""
time: str = self.data['commentedTime']
return time.removesuffix(' (edited)')
@property
def is_edited(self) -> bool:
"""
Whether or not the comment is edited.
### Note:
This property checks whether there is `'(edited)'` in the `commentedTime` property, because that's where you get that from.
See `Comments.Comment.commented_time`
"""
time: str = self.data['commentedTime']
return time.endswith('(edited)')
@property
def commentor_url(self) -> str:
"""
The URL of the channel that made the comment
"""
return self.data['commentorUrl']
@property
def replies_page(self) -> t.Optional[str]:
"""
Same as `Comments.nextpage`, but to load replies.
`None` means that there are no replies.
"""
return self.data['repliesPage']
@property
def hearted(self) -> bool:
"""
Whether or not the comment has been hearted
"""
return self.data['hearted']
@property
def like_count(self) -> int:
"""
The number of likes the comment has
"""
return self.data['likeCount']
@property
def pinned(self) -> bool:
"""
Whether or not the comment is pinned
"""
return self.data['pinned']
@property
def thumbnail(self) -> str:
"""
The thumbnail of the commentor's channel
"""
return self.data['thumbnail']
@property
def verified(self) -> bool:
"""
Whether or not the author of the comment is verified
"""
return self.data['verified']
def get_comments(self) -> t.List[Comment]:
"""
Obtain a list of comments
"""
return [self.Comment(comment_json) for comment_json in self.data['comments']]
def __iter__(self) -> t.Iterator[Comment]:
iter(self.get_comments())
@property
def disabled(self) -> bool:
"""
Whether or not the comments are disabled
"""
return self.data['disabled']
@property
def nextpage(self) -> t.Optional[str]:
"""
A JSON encoded page, which is used for the nextpage endpoint.
If there is no nextpage data, this returns `None`.
"""
return self.data['nextpage']

559
lib/piped_api/models/videos.py Normal file
View file

@ -0,0 +1,559 @@
import typing as t
from datetime import datetime, date, timedelta
from . import BasePipedModel
class Video(BasePipedModel):
@property
def title(self) -> str:
"""
The title/name of the video
"""
return self.data['title']
@property
def description(self) -> str:
"""
The description of the video
"""
return self.data['description']
@property
def upload_date(self) -> date:
"""
The date the video was uploaded at.
### Note:
Use `Video.data['uploadDate']` to get the raw string that was obtained from the API - this package
automatically converts the raw string to a `datetime.date` object.
"""
raw = self.data['uploadDate']
return datetime.strptime(raw, r"%Y-%m-%d").date()
@property
def uploader(self) -> str:
"""
The channel name of the author of the video
"""
return self.data['uploader']
@property
def uploader_url(self) -> str:
"""
The URI to the author's channel
"""
return self.data['uploaderUrl']
@property
def uploader_avatar(self) -> str:
"""
The URL to the video author's avatar image
"""
return self.data['uploaderAvatar']
@property
def thumbnail_url(self) -> str:
"""
The URL to the video's thumbnail image
"""
return self.data['thumbnail']
@property
def hls(self) -> t.Optional[str]:
"""
The hls manifest URL, to be used for Livestreams
"""
return self.data['hls']
@property
def dash(self) -> t.Optional[str]:
"""
The dash manifest URL for OTF streams
"""
return self.data['dash']
@property
def lbry_id(self) -> str:
"""
The lbry id of the video, if available. I assume this has something to do with https://lbry.com/
"""
return self.data['lbryId']
@property
def uploader_verified(self) -> str:
"""
Whether or not the channel that uploaded the video is verified by YouTube (badge)
"""
return self.data['uploaderVerified']
@property
def duration(self) -> timedelta:
"""
The duration of the video.
### Note:
The original value from the API was in seconds (`Video.data['duration']`), but this package
converts it to a `datetime.timedelta` object.
"""
return timedelta(seconds=self.data['duration'])
@property
def views(self) -> int:
"""
The number of views the video has received
"""
return self.data['views']
@property
def likes(self) -> int:
"""
The amount of likes the video has received. `-1` if hidden
"""
return self.data['likes']
@property
def dislikes(self) -> int:
"""
The amount of dislikes the video has received. `-1` if hidden
### Note:
This is obsolete since YouTube did a tiny gigantical little big whoopsie with their like system and screwed it all up
You can use awesome user-made projects such as https://returnyoutubedislike.com to obtain the dislike count
"""
return self.data['dislikes']
class Stream(BasePipedModel):
"""
Either an audio or video stream of a video
"""
@property
def url(self) -> str:
"""
The URL of the stream
"""
return self.data['url']
@property
def format(self) -> str:
"""
The format of the stream (`'M4A' or 'WEBMA_OPUS' or 'MPEG_4' or 'WEBM' or 'v3GPP'`
No, I don't know how many are there or what does each mean
"""
return self.data['format']
@property
def quality(self) -> str:
"""
The standard quality we all know and love (e. g.: `'240p'` for video or `'128k'` for audio)
"""
return self.data['quality']
@property
def mime_type(self) -> str:
"""
If you come from web development (or other invidious area that works with these French mimes),
then you already know what this is. If not, consider [checking the Mozilla documentation](https://developer.mozilla.org/en-US/docs/Web/HTTP/Basics_of_HTTP/MIME_types)
"""
return self.data['mimeType']
@property
def codec(self) -> str:
"""
What is this? I don't know. A codec?
"""
return self.data['codec']
@property
def video_only(self) -> bool:
"""
Whether or not the stream is video only (AKA. muted video)
"""
return self.data['videoOnly']
@property
def bitrate(self) -> int:
"""
The bitrate of the stream
"""
return self.data['bitrate']
@property
def init_start(self) -> int:
"""
Not sure what this does, but it seems to be useful for creating dash streams
"""
return self.data['initStart']
@property
def init_end(self) -> int:
"""
Not sure what this does, but it seems to be useful for creating dash streams
"""
return self.data['initEnd']
@property
def index_start(self) -> int:
"""
Not sure what this does, but it seems to be useful for creating dash streams
"""
return self.data['indexStart']
@property
def index_end(self) -> int:
"""
Not sure what this does, but it seems to be useful for creating dash streams
"""
return self.data['indexEnd']
@property
def width(self) -> int:
"""
The width of the stream. `'0'` for audio streams (makes sense)
"""
return self.data['width']
@property
def height(self) -> int:
"""
The height of the stream. `'0'` for audio streams (makes sense)
"""
return self.data['width']
@property
def fps(self) -> int:
"""
Frames Per Second. This is usually `'0'` for audio and `'30'` or `'60'` for video
"""
return self.data['fps']
def get_streams(self, type: t.Literal['video', 'audio']='video') -> t.List[Stream]:
"""
Get the streams of a video.
### Parameters:
- `type` - The type of stream to get. Either `'video'` or `'audio'`
"""
if type == 'video' or type == 'audio':
return [self.Stream(stream_data) for stream_data in self.data[f"{type}Streams"]]
raise ValueError('Invalid stream type. Must be either `video` or `audio`')
class RelatedStream(BasePipedModel):
"""
A related stream (e. g.: related video to the current one from the right sidebar, video related to/uploaded by a channel and trending video).
"""
@property
def url(self) -> str:
"""
The URL to the related video
"""
return self.data['url']
@property
def title(self) -> str:
"""
The title of the related video
"""
return self.data['title']
@property
def thumbnail(self) -> str:
"""
The thumbnail URL of the related video
"""
return self.data['thumbnail']
@property
def uploader_name(self) -> str:
"""
The name of the channel that uploaded the related video
"""
return self.data['uploaderName']
@property
def uploader_url(self) -> str:
"""
The URL of the channel that uploaded the related video
"""
return self.data['uploaderUrl']
@property
def uploader_avatar(self) -> str:
"""
The URL of the channel's avatar
"""
return self.data['uploaderAvatar']
@property
def uploaded_date(self) -> str:
"""
The date the related video was uploaded (format: `'x y ago'`)
"""
return self.data['uploadedDate']
@property
def short_description(self) -> t.Optional[str]:
"""
The short description of the related video. As far as I know, this is always `None` - perhaps some unused YouTube feature?
"""
return self.data['shortDescription']
@property
def duration(self) -> timedelta:
"""
The duration of the related video.
### Note:
The original value from the API was in seconds (`Video.data['duration']`), but this package
converts it to a `datetime.timedelta` object.
"""
return timedelta(seconds=self.data['duration'])
@property
def views(self) -> str:
"""
The amount of views the related video has received
"""
return self.data['views']
@property
def uploaded(self) -> datetime:
"""
The date the related video was uploaded (as a `datetime.datetime` object).
### Note:
The original value was in milliseconds since epoch (`Video.data['uploaded']`), but this package converts it to a `datetime.datetime` object.
"""
return datetime.fromtimestamp(self.data['uploaded'] / 1000)
@property
def uploader_verified(self) -> bool:
"""
Whether or not the channel that uploaded the related video is verified by YouTube (e. g.: has badge)
"""
return self.data['uploaderVerified']
@property
def related_videos(self) -> t.List[RelatedStream]:
"""
List of related streams
"""
return [self.RelatedStream(video_data) for video_data in self.data['relatedStreams']]
class Subtitle(BasePipedModel):
@property
def url(self) -> str:
"""
The URL to the subtitle
"""
return self.data['url']
@property
def mime_type(self) -> str:
"""
If you come from web development (or other invidious area that works with these French mimes),
then you already know what this is. If not, consider [checking the Mozilla documentation](https://developer.mozilla.org/en-US/docs/Web/HTTP/Basics_of_HTTP/MIME_types)
"""
return self.data['mimeType']
@property
def name(self) -> str:
"""
The name of the language the captions are in
"""
return self.data['name']
@property
def code(self) -> str:
"""
The country code for the captions
"""
return self.data['code']
@property
def auto_generated(self) -> bool:
"""
Whether or not the captions are auto-generated by YouTube
"""
return self.data['autoGenerated']
@property
def subtitles(self) -> t.List[Subtitle]:
"""
A list of captions for the video
"""
return [self.Subtitle(subtitle_data) for subtitle_data in self.data['subtitles']]
@property
def livestream(self) -> bool:
"""
Whether or not the video is a livestream
"""
return self.data['livestream']
@property
def proxy_url(self) -> str:
"""
The base URL for Piped proxy
"""
return self.data['proxyUrl']
class Chapter(BasePipedModel):
"""
A video chapter (or "section").
YouTube displays a list of chapters, if there are timestamps in the description.
"""
@property
def title(self) -> str:
"""
The title of the chapter
"""
return self.data['title']
@property
def image(self) -> str:
"""
The image URL for the chapter
"""
return self.data['image']
@property
def start(self) -> timedelta:
"""
The start time of the chapter
### Note:
The original value from the API was in seconds, this package automatically converts it to `datetime.timedelta`
"""
return timedelta(seconds=self.data['start'])
@property
def chapters(self) -> t.List[Chapter]:
"""
A list of chapters for the video
"""
return [self.Chapter(chapter_data) for chapter_data in self.data['chapters']]