[extractor/echo360] Implement all suggestions

yt_dlp/extractor/echo360.py

@@ -1,28 +1,20 @@
 import re
 
+from urllib.parse import urlparse, urlunparse
+
 from .common import InfoExtractor
 from ..utils import (
     determine_ext,
     float_or_none,
     traverse_obj,
-    variadic,
 )
 
 
 class Echo360IE(InfoExtractor):
-    _INSTANCES_RE = r'''(?:
+    _VALID_URL = r'''(?x)
-                            echo360\.ca|
+        https?://(?P<host>echo360\.(?:ca|net\.au|org|org\.au|org\.uk))/
-                            echo360\.net\.au|
+        media/(?P<id>[\da-fA-F]{8}-(?:[\da-fA-F]{4}-){3}[\da-fA-F]{12})/public
-                            echo360\.org\.au|
+    '''
-                            echo360\.org\.uk|
-                            echo360\.org|
-                        )'''
-    _UUID_RE = r'[\da-fA-F]{8}-[\da-fA-F]{4}-[\da-fA-F]{4}-[\da-fA-F]{4}-[\da-fA-F]{12}'
-    _VALID_URL = rf'''(?x)
-        https?://(?P<host>{_INSTANCES_RE})
-        /media/(?P<id>{_UUID_RE})/public'''
-
-    _API_BASE = 'https://%s/api/ui/echoplayer/public-links/%s/media/%s/player-properties'
 
     _TESTS = [
         {
@@ -49,21 +41,22 @@ class Echo360IE(InfoExtractor):
 
     def _call_api(self, host, video_id, media_id, session_token, **kwargs):
         return self._download_json(
-            self._API_BASE % (host, video_id, media_id), video_id,
+            f'https://{host}/api/ui/echoplayer/public-links/{video_id}/media/{media_id}/player-properties',
-            headers={'Authorization': f'Bearer {session_token}'}, **kwargs)
+            video_id, headers={'Authorization': f'Bearer {session_token}'}, **kwargs)
 
-    @staticmethod
+    def _replace_url_query(self, url, query_string):
-    def _update_url_query(uri, query_string):
         if query_string is not None:
-            return f'{uri.split("?", 1)[0]}?{query_string}'
+            return urlunparse(urlparse(url)._replace(query=query_string))
-        return uri
+        return url
 
-    @staticmethod
+    def _get_query_string(self, uri, query_strings):
-    def _get_query_string(uri, query_strings):
+        uri_base = urlparse(uri)._replace(query='', fragment='').geturl()
-        uri_base = uri.split("?", 1)[0]
         for query_string in query_strings:
+            try:
                 if re.match(query_string['uriPattern'], uri_base):
                     return query_string['queryString']
+            except re.error as re_error:
+                self.report_warning(f'Error in query string pattern `{re_error.pattern}`: {re_error.msg}')
         return None
 
     def _parse_mediapackage(self, video):
@@ -71,11 +64,11 @@ def _parse_mediapackage(self, video):
         query_strings = traverse_obj(video, ('sourceQueryStrings', 'queryStrings')) or []
 
         formats = []
-        for track in variadic(traverse_obj(video, ('playableAudioVideo', 'playableMedias')) or []):
+        for track in traverse_obj(video, ('playableAudioVideo', 'playableMedias', ...)):
             href = track.get('uri')
             if href is None:
                 continue
-            href = self._update_url_query(href, self._get_query_string(href, query_strings))
+            href = self._replace_url_query(href, self._get_query_string(href, query_strings))
             if track.get('isHls') or determine_ext(href, None) == 'm3u8':
                 hls_formats = self._extract_m3u8_formats(
                     href, video_id, live=track.get('isLive'), m3u8_id='hls',
@@ -85,7 +78,7 @@ def _parse_mediapackage(self, video):
                 for hls_format in hls_formats:
                     query_string = self._get_query_string(hls_format['url'], query_strings)
                     hls_format['extra_param_to_segment_url'] = query_string
-                    hls_format['url'] = self._update_url_query(hls_format['url'], query_string)
+                    hls_format['url'] = self._replace_url_query(hls_format['url'], query_string)
 
                 formats.extend(hls_formats)
 
@@ -94,24 +87,21 @@ def _parse_mediapackage(self, video):
             'formats': formats,
             'title': video.get('mediaName'),
             'duration': float_or_none(self._search_regex(
-                r'PT(\d+\.?\d+)S', traverse_obj(video, ('playableAudioVideo', 'duration')),
+                r'PT([\d.]+)S', traverse_obj(video, ('playableAudioVideo', 'duration')),
-                'video duration', default=None, fatal=False)),
+                'video duration', fatal=False)),
         }
 
     def _real_extract(self, url):
         host, video_id = self._match_valid_url(url).group('host', 'id')
         webpage = self._download_webpage(url, video_id)
 
-        player_config = self._search_json(r'Echo\["mediaPlayerBootstrapApp"\]\("', webpage,
+        player_config = self._search_json(
-                                          'player config', video_id,
+            r'Echo\["mediaPlayerBootstrapApp"\]\("', webpage, 'player config',
-                                          transform_source=lambda x: x.replace(R'\"', '"'))
+            video_id, transform_source=lambda x: x.replace(R'\"', '"'))
 
         urlh = self._request_webpage(
             f'https://{host}/api/ui/sessions/{player_config["sessionId"]}',
-            video_id,
+            video_id, 'Open video session', 'Unable to open video session')
-            note='Open video session',
-            errnote='Unable to open video session',
-        )
 
         return self._parse_mediapackage(self._call_api(
             host, player_config.get('shareLinkId') or player_config['publicLinkId'],