[`youtube`] Full plugin rewrite.

Instead of relying on the official Google YouTube API (limited, subject
to breaking changes with short/no notice depending on Google's strategy
against scrapers, and with an initial setup that has a high cost), we'll
just stick to Piped from now on.

It's free, it doesn't require API keys, it's unlikely to change, it's
not subject to Google's hostile practices against developers, and
anybody can run an instance.

platypush/plugins/media/search/youtube.py

@@ -1,9 +1,3 @@
-import re
-import urllib.parse
-import urllib.request
-
-import requests
-
 from platypush.context import get_plugin
 from platypush.plugins.media.search import MediaSearcher
 
@@ -16,67 +10,13 @@ class YoutubeMediaSearcher(MediaSearcher):
 
     def search(self, query: str, *_, **__):
         """
-        Performs a YouTube search either using the YouTube API (faster and
+        Performs a YouTube search using the ``youtube`` plugin.
-        recommended, it requires the :mod:`platypush.plugins.google.youtube`
-        plugin to be configured) or parsing the HTML search results (fallback
-        slower method)
         """
 
         self.logger.info('Searching YouTube for "%s"', query)
-
+        yt = get_plugin('youtube')
-        try:
+        assert yt, 'YouTube plugin not available/configured'
-            return self._youtube_search_api(query=query)
+        return yt.search(query=query).output
-        except Exception as e:
-            self.logger.warning(
-                (
-                    'Unable to load the YouTube plugin, '
-                    'falling back to HTML parse method: %s'
-                ),
-                e,
-            )
-
-        return self._youtube_search_html_parse(query=query)
-
-    @staticmethod
-    def _youtube_search_api(query):
-        yt = get_plugin('google.youtube')
-        assert yt, 'YouTube plugin not configured'
-        return [
-            {
-                'url': 'https://www.youtube.com/watch?v=' + item['id']['videoId'],
-                **item.get('snippet', {}),
-            }
-            for item in yt.search(query=query).output
-            if item.get('id', {}).get('kind') == 'youtube#video'
-        ]
-
-    def _youtube_search_html_parse(self, query):
-        query = urllib.parse.quote(query)
-        url = "https://www.youtube.com/results?search_query=" + query
-        html = requests.get(url, timeout=10).content
-        results = []
-
-        while html:
-            m = re.search(
-                r'(<a href="(/watch\?v=.+?)".+?yt-uix-tile-link.+?title="(.+?)".+?>)',
-                html,
-            )
-            if m:
-                results.append(
-                    {'url': 'https://www.youtube.com' + m.group(2), 'title': m.group(3)}
-                )
-
-                html = html.split(m.group(1))[1]
-            else:
-                html = ''
-
-        self.logger.info(
-            '%d YouTube video results for the search query "%s"',
-            len(results),
-            query,
-        )
-
-        return results
 
 
 # vim:sw=4:ts=4:et:

platypush/plugins/youtube/__init__.py

@@ -1,99 +1,77 @@
-from typing import Collection, Optional, Union
+import urllib.parse
-from platypush.plugins import action
+
-from platypush.plugins.google import GooglePlugin
+import requests
+
+from platypush.plugins import Plugin, action
 
 
-class YoutubePlugin(GooglePlugin):
+class YoutubePlugin(Plugin):
     r"""
     YouTube plugin.
 
-    Requirements:
+    Unlike other Google plugins, this plugin doesn't rely on the Google API.
 
-        1. Create your Google application, if you don't have one already, on
+    That's because the official YouTube API has been subject to many changes to
-           the `developers console <https://console.developers.google.com>`_.
+    prevent scraping, and it requires the user to tinker with the OAuth layer,
+    app permissions and app validation in order to get it working.
 
-        2. You may have to explicitly enable your user to use the app if the app
+    Instead, it relies on a `Piped <https://docs.piped.video/`_, an open-source
-           is created in test mode. Go to "OAuth consent screen" and add your user's
+    alternative YouTube gateway.
-           email address to the list of authorized users.
-
-        3. Select the scopes that you want to enable for your application, depending
-           on the integrations that you want to use.
-           See https://developers.google.com/identity/protocols/oauth2/scopes
-           for a list of the available scopes.
-
-        4. Click on "Credentials", then "Create credentials" -> "OAuth client ID".
-
-        5 Select "Desktop app", enter whichever name you like, and click "Create".
-
-        6. Click on the "Download JSON" icon next to your newly created client ID.
-
-        7. Generate a credentials file for the required scope:
-
-            .. code-block:: bash
-
-              mkdir -p <WORKDIR>/credentials/google
-              python -m platypush.plugins.google.credentials \
-                  'youtube.readonly' \
-                  <WORKDIR>/credentials/google/client_secret.json
 
+    It thus requires a link to a valid Piped instance.
     """
 
-    scopes = ['https://www.googleapis.com/auth/youtube.readonly']
+    def __init__(self, piped_api_url: str = 'https://pipedapi.kavin.rocks', **kwargs):
-
+        """
-    # See https://developers.google.com/youtube/v3/getting-started#part
+        :param piped_api_url: Base API URL of the Piped instance (default:
-    _default_parts = ['snippet']
+            ``https://pipedapi.kavin.rocks``).
-
+        """
-    # See https://developers.google.com/youtube/v3/getting-started#resources
+        super().__init__(**kwargs)
-    _default_types = ['video']
+        self._piped_api_url = piped_api_url
-
-    def __init__(self, *args, **kwargs):
-        super().__init__(scopes=self.scopes, *args, **kwargs)
 
     @action
-    def search(
+    def search(self, query: str, **_):
-        self,
-        parts: Optional[Union[str, Collection[str]]] = None,
-        query: str = '',
-        types: Optional[Union[str, Collection[str]]] = None,
-        max_results: int = 25,
-        **kwargs
-    ):
         """
-        Search for YouTube content.
+        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
+                }
 
-        :param parts: List of parts to get (default: snippet).
-            See the `Getting started - Part
-            <https://developers.google.com/youtube/v3/getting-started#part>`_.
-        :param query: Query string (default: empty string)
-        :param types: List of types to retrieve (default: video).
-            See the `Getting started - Resources
-            <https://developers.google.com/youtube/v3/getting-started#resources>`_.
-        :param max_results: Maximum number of items that will be returned (default: 25).
-        :param kwargs: Any extra arguments that will be transparently passed to the YouTube API.
-            See the `Getting started - parameters
-            <https://developers.google.com/youtube/v3/docs/search/list#parameters>`_.
-        :return: A list of YouTube resources.
-            See the `Getting started - Resource
-            <https://developers.google.com/youtube/v3/docs/search#resource>`_.
         """
+        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["title"],
+                "image": item["thumbnail"],
+                "duration": item["duration"],
+                "description": item["shortDescription"],
+            }
+            for item in rs.json().get("items", [])
+        ]
 
-        parts = parts or self._default_parts[:]
+        self.logger.info(
-        if isinstance(parts, list):
+            '%d YouTube video results for the search query "%s"',
-            parts = ','.join(parts)
+            len(results),
-
+            query,
-        types = types or self._default_types[:]
-        if isinstance(types, list):
-            types = ','.join(types)
-
-        service = self.get_service('youtube', 'v3')
-        result = (
-            service.search()
-            .list(part=parts, q=query, type=types, maxResults=max_results, **kwargs)
-            .execute()
         )
 
-        events = result.get('items', [])
+        return results
-        return events
 
 
 # vim:sw=4:ts=4:et: