diff --git a/modules/videos/downloader.py b/modules/videos/downloader.py index bf4c8ec3..223c38fb 100755 --- a/modules/videos/downloader.py +++ b/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: diff --git a/modules/videos/normalize_video_url.py b/modules/videos/normalize_video_url.py new file mode 100644 index 00000000..b87afff3 --- /dev/null +++ b/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 diff --git a/modules/videos/test/test_video_url_resolver.py b/modules/videos/test/test_video_url_resolver.py index d9b67777..73ab8670 100644 --- a/modules/videos/test/test_video_url_resolver.py +++ b/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 diff --git a/modules/videos/video_url_resolver.py b/modules/videos/video_url_resolver.py deleted file mode 100644 index b8e7a5fb..00000000 --- a/modules/videos/video_url_resolver.py +++ /dev/null @@ -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