From be28965d84df0415f4f5b81cfc0de0ae407cffb9 Mon Sep 17 00:00:00 2001 From: Fabio Manganiello Date: Mon, 13 Nov 2023 00:48:02 +0100 Subject: [PATCH] [`youtube`] Added `youtube.get_feed` action. --- platypush/plugins/youtube/__init__.py | 91 ++++++++++++---------- platypush/schemas/piped.py | 106 ++++++++++++++++++++++++++ 2 files changed, 158 insertions(+), 39 deletions(-) create mode 100644 platypush/schemas/piped.py diff --git a/platypush/plugins/youtube/__init__.py b/platypush/plugins/youtube/__init__.py index d82e6823..41e6e48a 100644 --- a/platypush/plugins/youtube/__init__.py +++ b/platypush/plugins/youtube/__init__.py @@ -1,9 +1,9 @@ -from datetime import datetime -import urllib.parse +from typing import Optional import requests from platypush.plugins import Plugin, action +from platypush.schemas.piped import PipedVideoSchema class YoutubePlugin(Plugin): @@ -22,13 +22,46 @@ class YoutubePlugin(Plugin): It thus requires a link to a valid Piped instance. """ - def __init__(self, piped_api_url: str = 'https://pipedapi.kavin.rocks', **kwargs): + _timeout = 20 + + def __init__( + self, + piped_api_url: str = 'https://pipedapi.kavin.rocks', + auth_token: Optional[str] = None, + **kwargs, + ): """ :param piped_api_url: Base API URL of the Piped instance (default: ``https://pipedapi.kavin.rocks``). + :param auth_token: Optional authentication token from the Piped + instance. This is required if you want to access your private feed + and playlists, but not for searching public videos. + + In order to retrieve an authentication token: + + 1. Login to your configured Piped instance. + 2. Copy the RSS/Atom feed URL on the _Feed_ tab. + 3. Copy the ``auth_token`` query parameter from the URL. + 4. Enter it in the ``auth_token`` field in the ``youtube`` section + of the configuration file. + """ super().__init__(**kwargs) self._piped_api_url = piped_api_url + self._auth_token = auth_token + + def _api_url(self, path: str = '') -> str: + return f"{self._piped_api_url}/{path}" + + def _request(self, path: str, auth: bool = True, **kwargs): + timeout = kwargs.pop('timeout', self._timeout) + if auth: + kwargs['params'] = kwargs.get('params', {}) + kwargs['params']['authToken'] = self._auth_token + + rs = requests.get(self._api_url(path), timeout=timeout, **kwargs) + rs.raise_for_status() + return rs.json() @action def search(self, query: str, **_): @@ -36,44 +69,11 @@ class YoutubePlugin(Plugin): Search for YouTube videos. :param query: Query string. - - :return: A list of results in the following format: - - .. code-block:: json - - { - "url": "https://www.youtube.com/watch?v=..." - "title": "Video title", - "description": "Video description", - "image": "https://i.ytimg.com/vi/.../hqdefault.jpg", - "duration": 300 - } - + :return: .. schema:: piped.PipedVideoSchema(many=True) """ self.logger.info('Searching YouTube for "%s"', query) - query = urllib.parse.quote(query) - url = f"{self._piped_api_url}/search?q=" + query + "&filter=all" - rs = requests.get(url, timeout=20) - rs.raise_for_status() - results = [ - { - "url": "https://www.youtube.com" + item["url"], - "title": item.get("title", '[No title]'), - "image": item.get("thumbnail"), - "duration": item.get("duration", 0), - "description": item.get("shortDescription"), - "channel": item.get("uploaderName"), - "channel_url": "https://www.youtube.com" + item["uploaderUrl"] - if item.get("uploaderUrl") - else None, - "channel_image": item.get("uploaderAvatar"), - "created_at": datetime.fromtimestamp(item["uploaded"] / 1000) - if item.get("uploaded") - else None, - } - for item in rs.json().get("items", []) - ] - + rs = self._request('search', auth=False, params={'q': query, 'filter': 'all'}) + results = PipedVideoSchema(many=True).dump(rs.get("items", [])) or [] self.logger.info( '%d YouTube video results for the search query "%s"', len(results), @@ -82,5 +82,18 @@ class YoutubePlugin(Plugin): return results + @action + def get_feed(self): + """ + Retrieve the YouTube feed. + + Depending on your account settings on the configured Piped instance, + this may return either the latest videos uploaded by your subscribed + channels, or the trending videos in the configured area. + + :return: .. schema:: piped.PipedVideoSchema(many=True) + """ + return PipedVideoSchema(many=True).dump(self._request('feed')) or [] + # vim:sw=4:ts=4:et: diff --git a/platypush/schemas/piped.py b/platypush/schemas/piped.py new file mode 100644 index 00000000..74e34095 --- /dev/null +++ b/platypush/schemas/piped.py @@ -0,0 +1,106 @@ +from datetime import datetime + +from marshmallow import EXCLUDE, fields, pre_dump +from marshmallow.schema import Schema + +from platypush.schemas import StrippedString + + +class PipedVideoSchema(Schema): + """ + Class for video items returned by the Piped API. + """ + + class Meta: + """ + Exclude unknown fields. + """ + + unknown = EXCLUDE + + url = fields.Url( + required=True, + metadata={ + 'description': 'Video URL', + 'example': 'https://youtube.com/watch?v=1234567890', + }, + ) + + title = StrippedString( + missing='[No Title]', + metadata={ + 'description': 'Video title', + 'example': 'My Video Title', + }, + ) + + image = fields.Url( + attribute='thumbnail', + metadata={ + 'description': 'Image URL', + 'example': 'https://i.ytimg.com/vi/1234567890/hqdefault.jpg', + }, + ) + + description = StrippedString( + attribute='shortDescription', + metadata={ + 'description': 'Video description', + 'example': 'My video description', + }, + ) + + duration = fields.Int( + missing=0, + metadata={ + 'description': 'Video duration in seconds', + 'example': 120, + }, + ) + + channel = fields.String( + attribute='uploaderName', + metadata={ + 'description': 'Channel name', + 'example': 'My Channel', + }, + ) + + channel_url = fields.Url( + attribute='uploaderUrl', + metadata={ + 'description': 'Channel URL', + 'example': 'https://youtube.com/channel/1234567890', + }, + ) + + channel_image = fields.Url( + attribute='uploaderAvatar', + metadata={ + 'description': 'Channel image URL', + 'example': 'https://i.ytimg.com/vi/1234567890/hqdefault.jpg', + }, + ) + + created_at = fields.DateTime( + attribute='uploaded', + metadata={ + 'description': 'Video upload date', + 'example': '2020-01-01T00:00:00', + }, + ) + + @pre_dump + def fill_urls(self, data: dict, **_): + for attr in ('url', 'uploaderUrl'): + if data.get(attr) and not data[attr].startswith('https://'): + data[attr] = f'https://youtube.com{data[attr]}' + + return data + + @pre_dump + def normalize_timestamps(self, data: dict, **_): + if data.get('uploaded') and isinstance(data['uploaded'], int): + data['uploaded'] = datetime.fromtimestamp(data["uploaded"] / 1000) + + return data