+
- str:
return f"{self._piped_api_url}/{path}"
- def _request(self, path: str, auth: bool = True, **kwargs):
+ def _request(
+ self, path: str, body: Optional[str] = None, auth: bool = True, **kwargs
+ ):
timeout = kwargs.pop('timeout', self._timeout)
if auth:
kwargs['params'] = kwargs.get('params', {})
@@ -61,10 +69,26 @@ class YoutubePlugin(Plugin):
kwargs['headers'] = kwargs.get('headers', {})
kwargs['headers']['Authorization'] = self._auth_token
+ if body:
+ kwargs['data'] = body
+
rs = requests.get(self._api_url(path), timeout=timeout, **kwargs)
rs.raise_for_status()
return rs.json()
+ @lru_cache(maxsize=10) # noqa
+ def _get_channel(self, id: str) -> dict: # pylint: disable=redefined-builtin
+ if (
+ id.startswith('http')
+ or id.startswith('https')
+ or id.startswith('/channel/')
+ ):
+ id = id.split('/')[-1]
+
+ return (
+ PipedChannelSchema().dump(self._request(f'channel/{id}')) or {} # type: ignore
+ )
+
@action
def search(self, query: str, **_) -> List[dict]:
"""
@@ -124,5 +148,53 @@ class YoutubePlugin(Plugin):
or []
)
+ @action
+ def get_subscriptions(self) -> List[dict]:
+ """
+ Retrieve the channels subscribed by the user logged in to the Piped
+ instance.
+
+ :return: .. schema:: piped.PipedChannelSchema(many=True)
+ """
+ return PipedChannelSchema(many=True).dump(self._request('subscriptions')) or []
+
+ @action
+ def get_channel(
+ self,
+ id: str, # pylint: disable=redefined-builtin
+ next_page_token: Optional[str] = None,
+ ) -> dict:
+ """
+ Retrieve the information and videos of a channel given its ID or URL.
+
+ :param id: Channel ID or URL.
+ :param next_page_token: Optional token to retrieve the next page of
+ results.
+ :return: .. schema:: piped.PipedChannelSchema
+ """
+ if (
+ id.startswith('http')
+ or id.startswith('https')
+ or id.startswith('/channel/')
+ ):
+ id = id.split('/')[-1]
+
+ info = {}
+ if next_page_token:
+ info = self._get_channel(id).copy()
+ info.pop('next_page_token', None)
+ info['items'] = []
+ next_page = base64.b64decode(next_page_token.encode()).decode()
+ response = {
+ **info,
+ **self._request(
+ f'nextpage/channel/{id}', params={'nextpage': next_page}, auth=False
+ ),
+ }
+ else:
+ response = self._request(f'channel/{id}')
+
+ return PipedChannelSchema().dump(response) or {} # type: ignore
+
# vim:sw=4:ts=4:et:
diff --git a/platypush/schemas/piped.py b/platypush/schemas/piped.py
index 96fcf449..98c28bf0 100644
--- a/platypush/schemas/piped.py
+++ b/platypush/schemas/piped.py
@@ -1,3 +1,4 @@
+import base64
from datetime import datetime
from marshmallow import EXCLUDE, fields, pre_dump
@@ -157,3 +158,109 @@ class PipedPlaylistSchema(Schema):
'example': 10,
},
)
+
+
+class PipedChannelSchema(Schema):
+ """
+ Class for channel items returned by the Piped API.
+ """
+
+ class Meta:
+ """
+ Exclude unknown fields.
+ """
+
+ unknown = EXCLUDE
+
+ id = fields.String(
+ required=True,
+ metadata={
+ 'description': 'Channel ID',
+ 'example': '1234567890',
+ },
+ )
+
+ url = fields.String(
+ required=True,
+ metadata={
+ 'description': 'Channel URL',
+ 'example': 'https://youtube.com/channel/1234567890',
+ },
+ )
+
+ name = StrippedString(
+ missing='[No Name]',
+ metadata={
+ 'description': 'Channel name',
+ 'example': 'My Channel Name',
+ },
+ )
+
+ description = StrippedString(
+ metadata={
+ 'description': 'Channel description',
+ 'example': 'My channel description',
+ },
+ )
+
+ image = fields.Url(
+ attribute='avatar',
+ metadata={
+ 'description': 'Channel image URL',
+ 'example': 'https://i.ytimg.com/vi/1234567890/hqdefault.jpg',
+ },
+ )
+
+ banner = fields.Url(
+ attribute='bannerUrl',
+ metadata={
+ 'description': 'Channel banner URL',
+ 'example': 'https://i.ytimg.com/vi/1234567890/hqdefault.jpg',
+ },
+ )
+
+ subscribers = fields.Int(
+ attribute='subscriberCount',
+ missing=0,
+ metadata={
+ 'description': 'Number of subscribers',
+ 'example': 1000,
+ },
+ )
+
+ next_page_token = fields.String(
+ attribute='nextpage',
+ metadata={
+ 'description': 'The token that should be passed to get the next page of results',
+ 'example': '1234567890',
+ },
+ )
+
+ items = fields.Nested(PipedVideoSchema, attribute='relatedStreams', many=True)
+
+ @pre_dump
+ def normalize_id_and_url(self, data: dict, **_):
+ if data.get('id'):
+ if not data.get('url'):
+ data['url'] = f'https://youtube.com/channel/{data["id"]}'
+ elif data.get('url'):
+ data['id'] = data['url'].split('/')[-1]
+ data['url'] = f'https://youtube.com{data["url"]}'
+ else:
+ raise AssertionError('Channel ID or URL not found')
+
+ return data
+
+ @pre_dump
+ def normalize_avatar(self, data: dict, **_):
+ if data.get('avatarUrl'):
+ data['avatar'] = data.pop('avatarUrl')
+
+ return data
+
+ @pre_dump
+ def serialize_next_page_token(self, data: dict, **_):
+ if data.get('nextpage'):
+ data['nextpage'] = base64.b64encode(data['nextpage'].encode()).decode()
+
+ return data