Removing extra parameters from video urls.

modules/videos/downloader.py

@@ -28,7 +28,7 @@ from .errors import UnknownChannel
 from .lib import get_downloader_config
 from .models import Video, Channel
 from .schema import ChannelPostRequest
-from .video_url_resolver import normalize_youtube_shorts_url
+from .normalize_video_url import normalize_video_url
 
 logger = logger.getChild(__name__)
 ydl_logger = logger.getChild('youtube-dl')
@@ -172,7 +172,7 @@ class ChannelDownloader(Downloader, ABC):
         downloads = [i.get('webpage_url') or i.get('url') for i in downloads]
 
         # YouTube Shorts are handled specially.
-        downloads = [normalize_youtube_shorts_url(i) for i in downloads]
+        downloads = [normalize_video_url(i) for i in downloads]
 
         # Only download those that have not yet been downloaded.
         already_downloaded = [i.url for i in video_downloader.already_downloaded(*downloads)]
@@ -205,7 +205,7 @@ class VideoDownloader(Downloader, ABC):
         if download.attempts >= 10:
             raise UnrecoverableDownloadError('Max download attempts reached')
 
-        url = download.url
+        url = normalize_video_url(download.url)
         info = download.info_json or extract_info(url)
 
         if not info:

modules/videos/normalize_video_url.py

@@ -0,0 +1,34 @@
+import re
+import urllib.parse
+
+from wrolpi.common import logger
+
+logger = logger.getChild(__name__)
+
+YOUTUBE_SHORT_REGEX = re.compile(r'https?://w{3}?\.?youtube.com/shorts/([-,._0-9a-zA-Z]{5,15})')
+YOUTU_BE_REGEX = re.compile(r'https?://(?:w{3}\.)?youtu\.be/([-,._0-9a-zA-Z]{5,15})')
+
+
+def strip_extra_from_url(url: str) -> str:
+    """Remove any query parameters which are unnecessary for a video url."""
+    if url.startswith('https://www.youtube.com/') and 'watch?' in url:
+        parsed = urllib.parse.urlparse(url)
+        query = urllib.parse.parse_qs(parsed.query)
+        if 'v' in query and ('list' in query or 'si' in query or 'pp' in query):
+            # Remove `list` or `si` from video URL.
+            v, = query['v']
+            url = f'https://www.youtube.com/watch?v={v}'
+    return url
+
+
+def normalize_video_url(url: str) -> str:
+    """Convert video URLs to what is expected to be in the yt-dlp info json."""
+    url = url.strip()
+    if match := YOUTUBE_SHORT_REGEX.match(url):
+        source_id = match.group(1)
+        url = f'https://www.youtube.com/watch?v={source_id}'
+    if match := YOUTU_BE_REGEX.match(url):
+        source_id = match.group(1)
+        url = f'https://www.youtube.com/watch?v={source_id}'
+    url = strip_extra_from_url(url)
+    return url

modules/videos/test/test_video_url_resolver.py

@@ -1,15 +1,22 @@
 import pytest
 
-from modules.videos.video_url_resolver import video_url_resolver
+from modules.videos.normalize_video_url import normalize_video_url
 
 
 @pytest.mark.parametrize(
-    'domain,entry,expected',
+    'url,expected',
     [
-        ('youtube.com', {'url': 'foo'}, 'https://youtube.com/watch?v=foo'),
-        ('youtube.com', {'webpage_url': 'https://youtube.com/watch?v=foo'}, 'https://youtube.com/watch?v=foo'),
-        ('youtube.com', {'url': 'https://youtube.com/watch?v=foo'}, 'https://youtube.com/watch?v=foo'),
+        ('https://example.com', 'https://example.com'),
+        ('https://www.youtube.com/watch?v=0xfMLNVFq2Y', 'https://www.youtube.com/watch?v=0xfMLNVFq2Y'),
+        ('https://www.youtube.com/shorts/0xfMLNVFq2Y', 'https://www.youtube.com/watch?v=0xfMLNVFq2Y'),
+        ('https://youtu.be/0xfMLNVFq2Y', 'https://www.youtube.com/watch?v=0xfMLNVFq2Y'),
+        ('https://youtu.be/0xfMLNVFq2Y', 'https://www.youtube.com/watch?v=0xfMLNVFq2Y'),
+        (
+                'https://www.youtube.com/watch?v=0xfMLNVFq2Y&list=PLMRyOtjhfFJUMPtB122lfeXebndQaePSF&index=2',
+                'https://www.youtube.com/watch?v=0xfMLNVFq2Y'
+        ),
+        ('https://youtu.be/0xfMLNVFq2Y?si=YALbJCpBgQcoY_Dp', 'https://www.youtube.com/watch?v=0xfMLNVFq2Y')
     ]
 )
-def test_video_url_resolver(domain, entry, expected):
-    assert video_url_resolver(domain, entry) == expected
+def test_normalize_video_url(url: str, expected: str):
+    assert normalize_video_url(url) == expected

modules/videos/video_url_resolver.py

@@ -1,41 +0,0 @@
-import re
-
-from wrolpi.common import logger
-
-logger = logger.getChild(__name__)
-
-formatters = {
-    'youtube.com': lambda i: f'https://youtube.com/watch?v={i["url"]}',
-    'www.youtube.com': lambda i: f'https://youtube.com/watch?v={i["url"]}',
-}
-
-
-def video_url_resolver(domain: str, entry: dict) -> str:
-    """Guess the full URL of a Youtube-DL entry."""
-    if 'webpage_url' in entry:
-        return entry['webpage_url']
-
-    if entry.get('url', '').startswith('https://'):
-        return entry['url']
-    if entry.get('url', '').startswith('http://'):
-        return entry['url']
-
-    try:
-        formatter = formatters[domain]
-        return formatter(entry)
-    except KeyError:
-        # Unable to resolve, try the URL as is.
-        return entry['url']
-
-
-YOUTUBE_SHORT_REGEX = re.compile(r'https?:\/\/w{3}?\.?youtube.com\/shorts\/(.{5,15})')
-
-
-def normalize_youtube_shorts_url(url: str) -> str:
-    """Convert a YouTube "shorts" URL to a regular video URL.
-
-    This is to avoid downloading shorts twice."""
-    if match := YOUTUBE_SHORT_REGEX.match(url.strip()):
-        source_id = match.group(1)
-        return f'https://www.youtube.com/watch?v={source_id}'
-    return url