From 1951ca548df0312ebddcc460cc0d4afe1be4e408 Mon Sep 17 00:00:00 2001 From: Roland Date: Thu, 7 May 2026 17:31:23 -0600 Subject: [PATCH] Using stubs to simplify tests. --- modules/videos/downloader.py | 167 +++++++++++++------------ modules/videos/test/test_downloader.py | 46 ++++--- wrolpi/downloader.py | 57 ++++++--- wrolpi/test/test_rss.py | 47 +++++++ 4 files changed, 196 insertions(+), 121 deletions(-) diff --git a/modules/videos/downloader.py b/modules/videos/downloader.py index 94731665..7102ef81 100755 --- a/modules/videos/downloader.py +++ b/modules/videos/downloader.py @@ -272,6 +272,91 @@ def extract_info(url: str, ydl: YoutubeDL = YDL, process: bool = False) -> dict: return ydl.extract_info(url, download=False, process=process) +def prepare_video_filename(url: str, out_dir: pathlib.Path, video_resolutions: str, video_format: str, + audio_only: bool = False, audio_format: str = 'mp3') \ + -> Tuple[pathlib.Path, dict]: + """Predict the on-disk path of a video/audio file from its URL using yt-dlp. + + Module-level (rather than a method on VideoDownloader) so tests can patch it once + and avoid the descriptor-binding pitfalls of patching `Class.method`. Returns the + absolute path the file will land at plus the yt-dlp entry dict for the URL. + """ + if not out_dir.is_dir(): + raise ValueError(f'Output directory does not exist! {out_dir=}') + + # YoutubeDL expects specific options, add onto the default options. + # Deep copy to avoid mutating the shared singleton config dict. + config = get_videos_downloader_config() + options = copy.deepcopy(config.yt_dlp_options) + + if audio_only: + options['format'] = 'bestaudio' + options['postprocessors'] = [{ + 'key': 'FFmpegExtractAudio', + 'preferredcodec': audio_format, + }] + # merge_output_format is not used for audio-only. + options.pop('merge_output_format', None) + else: + options['merge_output_format'] = video_format + options['format'] = video_resolutions + + # Convert WROLPi-specific variables to yt-dlp syntax + converted_format = convert_wrolpi_filename_format(config.file_name_format) + # yt-dlp expects the absolute path. + options['outtmpl'] = f'{out_dir}/{converted_format}' + options['cachdir'] = YTDLP_CACHE_DIR + + output_ext = audio_output_extension(audio_format) if audio_only else video_format + + # Create a new YoutubeDL for the output directory. + ydl = YoutubeDL(copy.deepcopy(options)) + ydl.params['logger'] = ydl_logger + ydl.add_default_info_extractors() + + # Get the path where the file will be saved. + try: + entry = extract_info(url, ydl=ydl, process=True) + # If the entry is a playlist (e.g. Twitter/X post with multiple videos), + # use the first entry for filename preparation. + if entry.get('entries'): + entries = list(entry['entries']) if not isinstance(entry['entries'], list) else entry['entries'] + if entries: + entry = entries[0] + final_filename = pathlib.Path(prepare_filename(entry, ydl=ydl)).absolute() + except DownloadError as e: + if ' Cannot write ' in str(e): + # yt-dlp does not handle long file names well, get the name from the error (lol) + last_line = str(e).splitlines()[-1] + full_path = last_line.split(' file ')[-1].strip() + if not full_path.startswith('/'): + logger.error(f'Failed to extract filename from {last_line}') + raise + else: + raise + + # Split filename from parent. + full_path = pathlib.Path(full_path) + parent = full_path.parent + filename, _ = split_path_stem_and_suffix(full_path.name) + + # Trim long filename, add output suffix, add back into parent directory. + filename = escape_file_name(filename) + final_filename = parent / f'{filename}.{output_ext}' + final_filename = trim_file_name(final_filename) + logger.debug(f'File name was too long. Trimmed to: {final_filename.name}') + + # Get entry info json. + options['outtmpl'] = str(final_filename) + ydl = YoutubeDL(copy.deepcopy(options)) + ydl.params['logger'] = ydl_logger + ydl.add_default_info_extractors() + entry = extract_info(url, ydl=ydl, process=True) + + logger.debug(f'Downloading {url} to {repr(str(final_filename))}') + return final_filename, entry + + def prepare_filename(entry: dict, ydl: YoutubeDL = YDL) -> str: """Get filename from YoutubeDL. Separated for testing.""" dir_name, file_name = os.path.split(ydl.prepare_filename(entry)) @@ -822,7 +907,7 @@ class VideoDownloader(Downloader, ABC): video_resolutions = effective['video_resolutions'] video_resolutions_str = ','.join(j for i in video_resolutions for j in VIDEO_RESOLUTION_MAP[i]) video_format = effective['video_format'] - video_path, entry = self.prepare_filename(url, out_dir, video_resolutions_str, video_format, + video_path, entry = prepare_video_filename(url, out_dir, video_resolutions_str, video_format, audio_only=audio_only, audio_format=audio_format) if settings.get('download_metadata_only'): @@ -1083,86 +1168,6 @@ class VideoDownloader(Downloader, ABC): logger.warning('Could not find channel') return None - @staticmethod - def prepare_filename(url: str, out_dir: pathlib.Path, video_resolutions: str, video_format: str, - audio_only: bool = False, audio_format: str = 'mp3') \ - -> Tuple[pathlib.Path, dict]: - """Get the full path of a video/audio file from its URL using yt-dlp.""" - if not out_dir.is_dir(): - raise ValueError(f'Output directory does not exist! {out_dir=}') - - # YoutubeDL expects specific options, add onto the default options. - # Deep copy to avoid mutating the shared singleton config dict. - config = get_videos_downloader_config() - options = copy.deepcopy(config.yt_dlp_options) - - if audio_only: - options['format'] = 'bestaudio' - options['postprocessors'] = [{ - 'key': 'FFmpegExtractAudio', - 'preferredcodec': audio_format, - }] - # merge_output_format is not used for audio-only. - options.pop('merge_output_format', None) - else: - options['merge_output_format'] = video_format - options['format'] = video_resolutions - - # Convert WROLPi-specific variables to yt-dlp syntax - converted_format = convert_wrolpi_filename_format(config.file_name_format) - # yt-dlp expects the absolute path. - options['outtmpl'] = f'{out_dir}/{converted_format}' - options['cachdir'] = YTDLP_CACHE_DIR - - output_ext = audio_output_extension(audio_format) if audio_only else video_format - - # Create a new YoutubeDL for the output directory. - ydl = YoutubeDL(copy.deepcopy(options)) - ydl.params['logger'] = ydl_logger - ydl.add_default_info_extractors() - - # Get the path where the file will be saved. - try: - entry = extract_info(url, ydl=ydl, process=True) - # If the entry is a playlist (e.g. Twitter/X post with multiple videos), - # use the first entry for filename preparation. - if entry.get('entries'): - entries = list(entry['entries']) if not isinstance(entry['entries'], list) else entry['entries'] - if entries: - entry = entries[0] - final_filename = pathlib.Path(prepare_filename(entry, ydl=ydl)).absolute() - except DownloadError as e: - if ' Cannot write ' in str(e): - # yt-dlp does not handle long file names well, get the name from the error (lol) - last_line = str(e).splitlines()[-1] - full_path = last_line.split(' file ')[-1].strip() - if not full_path.startswith('/'): - logger.error(f'Failed to extract filename from {last_line}') - raise - else: - raise - - # Split filename from parent. - full_path = pathlib.Path(full_path) - parent = full_path.parent - filename, _ = split_path_stem_and_suffix(full_path.name) - - # Trim long filename, add output suffix, add back into parent directory. - filename = escape_file_name(filename) - final_filename = parent / f'{filename}.{output_ext}' - final_filename = trim_file_name(final_filename) - logger.debug(f'File name was too long. Trimmed to: {final_filename.name}') - - # Get entry info json. - options['outtmpl'] = str(final_filename) - ydl = YoutubeDL(copy.deepcopy(options)) - ydl.params['logger'] = ydl_logger - ydl.add_default_info_extractors() - entry = extract_info(url, ydl=ydl, process=True) - - logger.debug(f'Downloading {url} to {repr(str(final_filename))}') - return final_filename, entry - @staticmethod async def download_info_json(download: Download, video_path: pathlib.Path) -> DownloadResult: """User has requested to refresh Video metadata. Download only metadata files to temporary directory. diff --git a/modules/videos/test/test_downloader.py b/modules/videos/test/test_downloader.py index 08c40d5d..c65fb69b 100644 --- a/modules/videos/test/test_downloader.py +++ b/modules/videos/test/test_downloader.py @@ -98,7 +98,7 @@ async def test_download_no_channel(test_session, video_download_manager, test_di url = 'https://www.youtube.com/watch?v=31jPEBiAC3c' mock_video_extract_info.return_value = info_json - with mock.patch('modules.videos.downloader.VideoDownloader.prepare_filename') as mock_prepare_filename: + with mock.patch('modules.videos.downloader.prepare_video_filename') as mock_prepare_filename: mock_prepare_filename.return_value = (video_path, {'id': 'foo'}) video_download_manager.create_download(test_session, url, video_downloader.name) await video_download_manager.wait_for_all_downloads() @@ -118,7 +118,7 @@ async def test_download_video_tags(test_session, video_download_manager, video_f url = 'https://www.youtube.com/watch?v=31jPEBiAC3c' mock_video_extract_info.return_value = copy(example_video_json) - with mock.patch('modules.videos.downloader.VideoDownloader.prepare_filename') as mock_prepare_filename: + with mock.patch('modules.videos.downloader.prepare_video_filename') as mock_prepare_filename: mock_prepare_filename.return_value = (video_path, {'id': 'foo'}) settings = dict(tag_names=[tag1.name, tag2.name]) video_download_manager.create_download(test_session, url, video_downloader.name, settings=settings) @@ -512,7 +512,7 @@ async def test_video_download(test_session, test_directory, mock_video_extract_i part_path.touch() url = 'https://www.youtube.com/watch?v=31jPEBiAC3c' - with mock.patch('modules.videos.downloader.VideoDownloader.prepare_filename') as mock_prepare_filename: + with mock.patch('modules.videos.downloader.prepare_video_filename') as mock_prepare_filename: mock_video_extract_info.return_value = example_video_json mock_prepare_filename.return_value = (video_path, {'id': 'foo'}) @@ -549,7 +549,7 @@ async def test_download_result(test_session, test_directory, video_download_mana image_file.rename(video_file.with_suffix('.jpg')) mock_video_extract_info.return_value = example_video_json - with mock.patch('modules.videos.downloader.VideoDownloader.prepare_filename') as mock_prepare_filename: + with mock.patch('modules.videos.downloader.prepare_video_filename') as mock_prepare_filename: mock_prepare_filename.return_value = [video_file, {'id': 'foo'}] video_download_manager.create_download(test_session, 'https://example.com', video_downloader.name) await video_download_manager.wait_for_all_downloads() @@ -670,13 +670,12 @@ async def test_download_playlist(test_session, test_directory, mock_video_extrac mock_video_extract_info.return_value = example_playlist_json # Playlist info is fetched first. - with mock.patch('modules.videos.downloader.VideoDownloader.execute_download') as mock_video_do_download: - mock_video_do_download.return_value = DownloadResult(success=True) # Don't download the videos. - # Drive the phase-split API directly: prepare → execute → finalize. - from modules.archive.conftest import make_test_ctx - prepared = channel_downloader.prepare_download(test_session, download) - executed = await channel_downloader.execute_download(prepared, make_test_ctx(), download=download) - result = channel_downloader.finalize_download(test_session, download, executed) + # Drive the phase-split API directly: prepare → execute → finalize. No dispatch + # loop, so child video downloads are never spawned here — no need to stub them. + from modules.archive.conftest import make_test_ctx + prepared = channel_downloader.prepare_download(test_session, download) + executed = await channel_downloader.execute_download(prepared, make_test_ctx(), download=download) + result = channel_downloader.finalize_download(test_session, download, executed) assert result.success is True, 'Download was not successful' assert set(result.downloads) == { 'https://www.youtube.com/watch?v=video_2_url', # Shorts is converted to regular URL. @@ -904,11 +903,12 @@ async def test_channel_downloader_sets_collection_id(test_session, channel_facto } mock_video_extract_info.return_value = mock_info - # Mock Channel.refresh_files and file_worker.wait_for_job to avoid background tasks - with mock.patch('modules.videos.downloader.VideoDownloader.execute_download') as mock_video_do_download, \ - mock.patch.object(Channel, 'refresh_files', return_value='test-job-id'), \ + # Drive the phase-split API directly — no dispatch loop fires, so child video + # downloads are never spawned and don't need stubbing. refresh_files and + # file_worker.wait_for_job are still patched because ChannelDownloader's + # prepare_channel_for_downloads calls them inline. + with mock.patch.object(Channel, 'refresh_files', return_value='test-job-id'), \ mock.patch('modules.videos.downloader.file_worker.wait_for_job', new_callable=mock.AsyncMock): - mock_video_do_download.return_value = DownloadResult(success=True) from modules.archive.conftest import make_test_ctx prepared = channel_downloader.prepare_download(test_session, download) executed = await channel_downloader.execute_download(prepared, make_test_ctx(), download=download) @@ -956,10 +956,9 @@ async def test_channel_passes_all_inheritable_settings_to_video(test_session, ch } mock_video_extract_info.return_value = mock_info - with mock.patch('modules.videos.downloader.VideoDownloader.execute_download') as mock_video_do_download, \ - mock.patch.object(Channel, 'refresh_files', return_value='test-job-id'), \ + # Direct phase-split call — child video downloads are never dispatched here. + with mock.patch.object(Channel, 'refresh_files', return_value='test-job-id'), \ mock.patch('modules.videos.downloader.file_worker.wait_for_job', new_callable=mock.AsyncMock): - mock_video_do_download.return_value = DownloadResult(success=True) from modules.archive.conftest import make_test_ctx prepared = channel_downloader.prepare_download(test_session, download) executed = await channel_downloader.execute_download(prepared, make_test_ctx(), download=download) @@ -1003,10 +1002,9 @@ async def test_channel_does_not_pass_continue_dl_nooverwrites(test_session, chan } mock_video_extract_info.return_value = mock_info - with mock.patch('modules.videos.downloader.VideoDownloader.execute_download') as mock_video_do_download, \ - mock.patch.object(Channel, 'refresh_files', return_value='test-job-id'), \ + # Direct phase-split call — child video downloads are never dispatched here. + with mock.patch.object(Channel, 'refresh_files', return_value='test-job-id'), \ mock.patch('modules.videos.downloader.file_worker.wait_for_job', new_callable=mock.AsyncMock): - mock_video_do_download.return_value = DownloadResult(success=True) from modules.archive.conftest import make_test_ctx prepared = channel_downloader.prepare_download(test_session, download) executed = await channel_downloader.execute_download(prepared, make_test_ctx(), download=download) @@ -1035,7 +1033,7 @@ async def test_video_download_uses_effective_settings(test_session, test_directo # Download with writesubtitles=False override (global default is True) settings = {'writesubtitles': False, 'sleep_requests': 5.0} - with mock.patch('modules.videos.downloader.VideoDownloader.prepare_filename') as mock_prepare_filename: + with mock.patch('modules.videos.downloader.prepare_video_filename') as mock_prepare_filename: mock_video_extract_info.return_value = example_video_json mock_prepare_filename.return_value = (video_path, {'id': 'foo'}) @@ -1073,7 +1071,7 @@ async def test_live_video_retry_seconds(test_session, test_directory, mock_video url = 'https://www.youtube.com/watch?v=live_video' mock_video_extract_info.return_value = example_video_json - with mock.patch('modules.videos.downloader.VideoDownloader.prepare_filename') as mock_prepare_filename: + with mock.patch('modules.videos.downloader.prepare_video_filename') as mock_prepare_filename: # Return a path that won't exist (video wasn't downloaded because it's live). mock_prepare_filename.return_value = (simple_channel.directory / 'live video.mp4', {'id': 'foo'}) @@ -1134,7 +1132,7 @@ async def test_video_download_playlist_appended_extension(test_session, test_dir url = 'https://x.com/example/status/123' mock_video_extract_info.return_value = example_video_json - with mock.patch('modules.videos.downloader.VideoDownloader.prepare_filename') as mock_prepare_filename: + with mock.patch('modules.videos.downloader.prepare_video_filename') as mock_prepare_filename: # prepare_filename returns the path without the real extension (ending in .NA). bogus_path = video_path_na.parent / video_path_na.stem # Removes .mp4, leaving .NA as "extension" mock_prepare_filename.return_value = (bogus_path, {'id': 'foo'}) diff --git a/wrolpi/downloader.py b/wrolpi/downloader.py index 446190f4..1e80bd9c 100644 --- a/wrolpi/downloader.py +++ b/wrolpi/downloader.py @@ -2143,13 +2143,39 @@ class RSSDownloader(Downloader): # (e.g., youtube.com/watch?v=ABC123) to match what is stored in the database. from modules.videos.normalize_video_url import normalize_video_url urls = [normalize_video_url(i) for i in urls] - # Duration filtering performs network I/O (yt-dlp duration probe) — must run - # in execute, not finalize. This may probe URLs that DB-dedupe would have - # eliminated; fetch_video_duration is cached, so duplicate probes are cheap. + + # Pre-filter against DB + skip-list BEFORE duration filtering: each duration + # probe sleeps random.randint(1, 3) seconds inside filter_videos, so checking + # the DB first avoids dozens of unnecessary sleeps for feeds where most videos + # are already downloaded. Mirrors the legacy filter ordering. The session is + # opened locally and closed before the long-running filter_videos await — the + # manager's outer session is not held across this work. + urls = self._filter_already_known(prepared.sub_downloader, urls) + + # Duration filtering performs network I/O (yt-dlp duration probe). Now every + # probe is on a genuinely new URL. urls = await self.filter_videos(log_stub, urls) return ExecutedRSS(yt_channel_id=yt_channel_id, candidate_urls=urls) + @staticmethod + def _filter_already_known(sub_downloader: 'Downloader', urls: List[str]) -> List[str]: + """Drop URLs already downloaded, already queued, or on the skip list. + + Opens a short-lived self-managed session for the DB queries; the manager's + outer session is not held across this call. + """ + if not urls: + return urls + with get_db_session() as session: + already = sub_downloader.already_downloaded(session, *urls) + already_urls = {a.url for a in already} + urls = [u for u in urls if u not in already_urls] + existing_download_urls = {d[0] for d in session.query(Download.url).filter(Download.url.in_(urls))} + urls = [u for u in urls if u not in existing_download_urls] + urls = [u for u in urls if not download_manager.is_skipped(u)] + return urls + def finalize_download(self, session: Session, download: Download, executed: ExecutedRSS) -> DownloadResult: if executed.error: @@ -2159,21 +2185,20 @@ class RSSDownloader(Downloader): if executed.yt_channel_id and not (download.location or download.collection_id): self.apply_yt_channel(session, download, executed.yt_channel_id) - # Resolve sub_downloader via the manager (matches prepare; cheap dict lookup). - sub_downloader = download_manager.find_downloader_by_name(download.sub_downloader) urls = executed.candidate_urls - # DB dedupe: drop URLs already downloaded under this sub_downloader, plus URLs - # already queued in the Download table. - if sub_downloader and urls: - already = sub_downloader.already_downloaded(session, *urls) - already_urls = {a.url for a in already} - urls = [u for u in urls if u not in already_urls] - existing_download_urls = {d[0] for d in session.query(Download.url).filter(Download.url.in_(urls))} - urls = [u for u in urls if u not in existing_download_urls] - - # Skip-list filter (manager-owned shared_ctx state, not DB). - urls = [u for u in urls if not download_manager.is_skipped(u)] + # Video URLs were already deduped in execute_download (pre-duration-filter, to + # avoid wasted sleeps). Non-video sub_downloaders skip the in-execute filter + # because they don't run filter_videos, so dedupe them here. + if download.sub_downloader != 'video': + sub_downloader = download_manager.find_downloader_by_name(download.sub_downloader) + if sub_downloader and urls: + already = sub_downloader.already_downloaded(session, *urls) + already_urls = {a.url for a in already} + urls = [u for u in urls if u not in already_urls] + existing_download_urls = {d[0] for d in session.query(Download.url).filter(Download.url.in_(urls))} + urls = [u for u in urls if u not in existing_download_urls] + urls = [u for u in urls if not download_manager.is_skipped(u)] logger.info(f'Successfully got {len(urls)} new URLs from RSS {download.url}') diff --git a/wrolpi/test/test_rss.py b/wrolpi/test/test_rss.py index df6eea2e..e8f722e9 100644 --- a/wrolpi/test/test_rss.py +++ b/wrolpi/test/test_rss.py @@ -480,3 +480,50 @@ def test_finalize_passes_destination_from_column(test_session, test_download_man assert result.success is True assert result.settings['destination'].endswith('archive/custom/path') + + +@pytest.mark.asyncio +async def test_execute_dedupes_before_filter_videos(test_session, video_factory): + """Regression: filter_videos must see only URLs that aren't already downloaded. + + Each call inside filter_videos sleeps 1-3s per URL; running it on already-downloaded + videos is wasted work. The legacy code did dedupe BEFORE the duration filter, and + the phase-split must preserve that ordering. + + A spy on filter_videos captures exactly what reaches it. + """ + from modules.videos.downloader import video_downloader + from wrolpi.downloader import rss_downloader + + # One video already in the DB; the other URL is new. + v = video_factory(source_id='ABC123') + v.file_group.url = 'https://www.youtube.com/watch?v=ABC123' + test_session.commit() + + seen_by_filter = [] + + async def spy_filter_videos(_download, urls): + seen_by_filter.extend(urls) + return urls + + # filter_videos is @staticmethod; patching with a plain function rebinds self, + # so wrap in staticmethod() to preserve the original calling convention. + spy = staticmethod(spy_filter_videos) + + prepared = PreparedRSS( + url='https://example.com/feed', + sub_downloader_name='video', + sub_downloader=video_downloader, + settings={'maximum_duration': 600}, # any min/max triggers filter_videos + ) + + feed = dict(bozo=0, entries=[ + dict(link='https://www.youtube.com/shorts/ABC123'), # already in DB + dict(link='https://www.youtube.com/shorts/NEW789'), # new + ]) + + with mock.patch('wrolpi.downloader.parse_feed', return_value=feed), \ + mock.patch.object(RSSDownloader, 'filter_videos', spy): + await rss_downloader.execute_download(prepared, make_test_ctx()) + + assert seen_by_filter == ['https://www.youtube.com/watch?v=NEW789']