diff --git a/learning_resources/etl/loaders.py b/learning_resources/etl/loaders.py index 0460777b29..27d9c65e53 100644 --- a/learning_resources/etl/loaders.py +++ b/learning_resources/etl/loaders.py @@ -8,7 +8,7 @@ from django.contrib.auth import get_user_model from django.core.cache import caches from django.db import transaction -from django.db.models import Max, Q +from django.db.models import Max, Q, QuerySet from learning_resources.constants import ( CONTENT_TYPE_PAGE, @@ -1870,130 +1870,123 @@ def load_videos_from_content_files( return videos -def load_playlists( - video_channel: VideoChannel, playlists_data: iter -) -> list[LearningResource]: +def upsert_video_channel(video_channel_data: dict) -> VideoChannel: """ - Load a list of video playlists into the database - - Args: - video_channel (VideoChannel): the video channel instance this playlist is under - playlists_data (iter of dict): iterable of the video playlists + Create or update the VideoChannel row itself, without touching its playlists + Arg: + video_channel_data (dict): + the normalized video channel data, without a "playlists" key Returns: - list of LearningResource: - the created or updated LearningResources for the playlists + VideoChannel: the updated or created video channel """ - playlists = [ - playlist - for playlist in ( - load_playlist(video_channel, playlist_data) - for playlist_data in playlists_data - ) - if playlist is not None - ] - playlist_ids = [playlist.id for playlist in playlists] + channel_data = { + **video_channel_data, + "etl_source": ETLSource.youtube.name, + "published": True, + } + channel_id = channel_data.pop("channel_id") + video_channel, _ = VideoChannel.objects.select_for_update().update_or_create( + channel_id=channel_id, defaults=channel_data + ) + return video_channel - # remove playlists that no longer exist - playlists_to_unpublish = LearningResource.objects.filter( - video_playlist__channel=video_channel - ).exclude(id__in=playlist_ids) - playlists_to_unpublish.update(published=False) - bulk_resources_unpublished_actions( - playlists_to_unpublish.values_list("id", flat=True), - LearningResourceType.video_playlist.name, +def unpublish_orphaned_videos(playlist_ids: list[int] | None = None) -> None: + """ + Unpublish published videos that are in no published playlist. + + Args: + playlist_ids (list of int or None): only consider the videos of these + playlist resources; if None, sweep every playlist video + """ + playlist_videos = LearningResourceRelationship.objects.filter( + relation_type=LearningResourceRelationTypes.PLAYLIST_VIDEOS.value + ) + candidates = ( + playlist_videos.filter(parent_id__in=playlist_ids) + if playlist_ids is not None + else playlist_videos ) - return playlists + orphaned_video_ids = list( + LearningResource.objects.filter( + published=True, id__in=candidates.values("child_id") + ) + .exclude( + # a video in another, still-published playlist isn't orphaned + id__in=playlist_videos.filter(parent__published=True).values("child_id") + ) + .values_list("id", flat=True) + ) + + if orphaned_video_ids: + LearningResource.objects.filter(id__in=orphaned_video_ids).update( + published=False + ) + bulk_resources_unpublished_actions( + orphaned_video_ids, LearningResourceType.video.name + ) -def load_video_channel(video_channel_data: dict) -> VideoChannel: +def unpublish_playlists(playlist_resources: QuerySet) -> None: """ - Load a single video channel into the database + Unpublish playlist resources, and any video they leave orphaned - Arg: - video_channel_data (dict): - the normalized video channel data - Returns: - VideoChannel: the updated or created video channel + Args: + playlist_resources (QuerySet): the playlist LearningResources to unpublish """ - channel_id = video_channel_data.pop("channel_id") - playlists_data = video_channel_data.pop("playlists", []) + unpublished_ids = list(playlist_resources.values_list("id", flat=True)) + if not unpublished_ids: + return - video_channel, _ = VideoChannel.objects.select_for_update().update_or_create( - channel_id=channel_id, defaults=video_channel_data + LearningResource.objects.filter(id__in=unpublished_ids).update(published=False) + bulk_resources_unpublished_actions( + unpublished_ids, LearningResourceType.video_playlist.name ) - load_playlists(video_channel, playlists_data) - - return video_channel + unpublish_orphaned_videos(unpublished_ids) -def load_youtube_video_channels(video_channels_data: iter) -> list[VideoChannel]: +def unpublish_removed_playlists( + video_channel: VideoChannel, playlist_ids: list[str] +) -> None: """ - Load a list of video channels + Unpublish the channel's playlists that are no longer in its youtube listing Args: - video_channels_data (iter of dict): iterable of the video channels data + video_channel (VideoChannel): the video channel + playlist_ids (list of str): youtube ids of the channel's current playlists + """ + unpublish_playlists( + LearningResource.objects.filter(video_playlist__channel=video_channel).exclude( + readable_id__in=playlist_ids + ) + ) - Returns: - list of VideoChannel: the loaded video channels - """ - video_channels = [] - channel_ids = [] - for video_channel_data in video_channels_data: - channel_id = video_channel_data["channel_id"] - channel_ids.append(channel_id) - video_channel_data["etl_source"] = ETLSource.youtube.name - video_channel_data["published"] = True - try: - video_channel = load_video_channel(video_channel_data) - except ExtractException: - # video_channel_data has lazily evaluated generators, - # one of them could raise an extraction error - # this is a small pollution of separation of concerns - # but this allows us to stream the extracted data w/ generators - # as opposed to having to load everything into memory, - # which will eventually fail - log.exception( - "Error with extracted video channel: channel_id=%s", channel_id - ) - else: - video_channels.append(video_channel) +def unpublish_removed_youtube_channels(channel_ids: list[str]) -> None: + """ + Unpublish everything youtube no longer offers under the configured channels. + + Channels that aren't in channel_ids are unpublished, along with their + playlists and any video left without a published playlist. + + Args: + channel_ids (list of str): youtube ids of the configured channels + """ VideoChannel.objects.filter(etl_source=ETLSource.youtube.name).exclude( channel_id__in=channel_ids ).update(published=False) # Unpublish any video playlists not included in published channels - orphaned_playlist_ids = ( - VideoPlaylist.objects.exclude(channel__channel_id__in=channel_ids) - .filter(channel__etl_source=ETLSource.youtube.name) - .values_list("learning_resource__id", flat=True) - ) - - if orphaned_playlist_ids: - LearningResource.objects.filter(id__in=orphaned_playlist_ids).update( - published=False - ) - bulk_resources_unpublished_actions( - orphaned_playlist_ids, LearningResourceType.video_playlist.name + unpublish_playlists( + LearningResource.objects.filter( + id__in=VideoPlaylist.objects.exclude(channel__channel_id__in=channel_ids) + .filter(channel__etl_source=ETLSource.youtube.name) + .values_list("learning_resource__id", flat=True) ) - - # Unpublish any published videos that aren't in at least one published playlist - orphaned_video_ids = ( - LearningResourceRelationship.objects.filter( - relation_type=LearningResourceRelationTypes.PLAYLIST_VIDEOS.value - ) - .exclude(parent__published=True) - .values_list("child", flat=True) ) - if orphaned_video_ids: - LearningResource.objects.filter(id__in=orphaned_video_ids).update( - published=False - ) - bulk_resources_unpublished_actions( - orphaned_video_ids, LearningResourceType.video.name - ) - return video_channels + # Backstop: catch videos orphaned by anything the per-playlist path missed, + # such as a run that died partway through or a playlist load that bailed + unpublish_orphaned_videos() diff --git a/learning_resources/etl/loaders_test.py b/learning_resources/etl/loaders_test.py index fa8c962b4a..c5e806ad7b 100644 --- a/learning_resources/etl/loaders_test.py +++ b/learning_resources/etl/loaders_test.py @@ -33,7 +33,6 @@ ProgramLoaderConfig, ) from learning_resources.etl.edx_shared import sync_edx_course_files -from learning_resources.etl.exceptions import ExtractException from learning_resources.etl.loaders import ( ProgramLoadResult, calculate_completeness, @@ -47,7 +46,6 @@ load_ovs_playlist, load_ovs_playlists, load_playlist, - load_playlists, load_podcast, load_podcast_episode, load_podcasts, @@ -62,7 +60,9 @@ load_video_with_content_file, load_videos, load_videos_from_content_files, - load_youtube_video_channels, + unpublish_orphaned_videos, + unpublish_removed_playlists, + unpublish_removed_youtube_channels, ) from learning_resources.etl.mitxonline import transform_programs from learning_resources.etl.utils import get_s3_prefix_for_source @@ -100,8 +100,6 @@ Program, TutorProblemFile, Video, - VideoChannel, - VideoPlaylist, ) from learning_resources.test_utils import set_up_topics from main.utils import now_in_utc @@ -2814,57 +2812,78 @@ def test_load_videos_from_content_files_empty_input(): assert result == [] -def test_load_playlists_unpublish(mocker): - """Test load_playlists when a video/playlist gets unpublished""" - mocker.patch("learning_resources_search.tasks.bulk_deindex_learning_resources.si") +def _add_playlist_videos(playlist_resource, videos): + """Attach videos to a playlist resource""" + playlist_resource.resources.set( + videos, + through_defaults={ + "relation_type": LearningResourceRelationTypes.PLAYLIST_VIDEOS.value + }, + ) + + +def test_unpublish_removed_playlists(mock_upsert_tasks): + """A removed playlist should take the videos it orphans with it""" + channel = VideoChannelFactory.create() + kept, removed = VideoPlaylistFactory.create_batch(2, channel=channel) + other_channel_playlist = VideoPlaylistFactory.create() + + orphaned_video = VideoFactory.create().learning_resource + shared_video = VideoFactory.create().learning_resource + _add_playlist_videos(removed.learning_resource, [orphaned_video, shared_video]) + _add_playlist_videos(kept.learning_resource, [shared_video]) + + unpublish_removed_playlists(channel, [kept.learning_resource.readable_id]) + + for playlist, expected in ( + (kept, True), + (removed, False), + (other_channel_playlist, True), + ): + playlist.refresh_from_db() + assert playlist.learning_resource.published is expected + + orphaned_video.refresh_from_db() + assert orphaned_video.published is False + # still listed under a published playlist, so not orphaned + shared_video.refresh_from_db() + assert shared_video.published is True + + +def test_unpublish_removed_playlists_noop(mocker): + """Nothing should be unpublished when the channel's playlists all still exist""" mock_bulk_unpublish = mocker.patch( "learning_resources.etl.loaders.bulk_resources_unpublished_actions", ) channel = VideoChannelFactory.create() - - playlists = sorted( - VideoPlaylistFactory.create_batch(4, channel=channel), - key=lambda playlist: playlist.id, + playlists = VideoPlaylistFactory.create_batch(2, channel=channel) + _add_playlist_videos( + playlists[0].learning_resource, [VideoFactory.create().learning_resource] ) - playlist_id = playlists[0].learning_resource.readable_id - playlist_title = playlists[0].learning_resource.title - assert playlists[0].learning_resource.published is True - playlists_data = [ - { - "playlist_id": playlist_id, - "url": f"https://youtube.com/playlist?list={playlist_id}", - "image": { - "url": f"https://i.ytimg.com/vi/{playlist_id}/hqdefault.jpg", - "alt": playlist_title, - }, - "published": True, - "videos": [], - } - ] - load_playlists(channel, playlists_data) - assert ( - LearningResource.objects.filter( - resource_type="video_playlist", published=True - ).count() - == 1 + unpublish_removed_playlists( + channel, [playlist.learning_resource.readable_id for playlist in playlists] ) - for playlist in playlists: - playlist.refresh_from_db() - if playlist.id == playlists[0].id: - assert playlist.learning_resource.published is True - else: - assert playlist.learning_resource.published is False + mock_bulk_unpublish.assert_not_called() - expected_unpublished_ids = sorted(p.learning_resource.id for p in playlists[1:]) - playlist_unpublish_call = next( - call - for call in mock_bulk_unpublish.call_args_list - if call[0][1] == LearningResourceType.video_playlist.name - ) - actual_unpublished_ids = sorted(playlist_unpublish_call[0][0]) - assert actual_unpublished_ids == expected_unpublished_ids + +def test_unpublish_orphaned_videos_sweeps_everything(mock_upsert_tasks): + """Called without playlist ids it should catch any video left unlisted""" + published_playlist = VideoPlaylistFactory.create().learning_resource + stale_playlist = VideoPlaylistFactory.create(is_unpublished=True).learning_resource + + orphaned_video = VideoFactory.create().learning_resource + listed_video = VideoFactory.create().learning_resource + _add_playlist_videos(stale_playlist, [orphaned_video]) + _add_playlist_videos(published_playlist, [listed_video]) + + unpublish_orphaned_videos() + + orphaned_video.refresh_from_db() + assert orphaned_video.published is False + listed_video.refresh_from_db() + assert listed_video.published is True @pytest.mark.parametrize("playlist_exists", [True, False]) @@ -3107,84 +3126,21 @@ def test_load_ovs_playlists_empty_aborts(mocker): assert vp.learning_resource.published is True -def test_load_youtube_video_channels(): - """Test load_youtube_video_channels""" - assert VideoChannel.objects.count() == 0 - assert VideoPlaylist.objects.count() == 0 - - channels_data = [] - for channel in VideoChannelFactory.build_batch(3): - channel_data = model_to_dict(channel) - - playlist = VideoPlaylistFactory.build() - playlist_data = model_to_dict(playlist) - playlist_id = playlist.learning_resource.readable_id - playlist_data["playlist_id"] = playlist_id - playlist_data["url"] = f"https://youtube.com/playlist?list={playlist_id}" - playlist_data["image"] = { - "url": f"https://i.ytimg.com/vi/{playlist_id}/hqdefault.jpg", - "alt": playlist.learning_resource.title, - } - del playlist_data["id"] - del playlist_data["channel"] - del playlist_data["learning_resource"] - del playlist_data["parent_learning_resource"] - - channel_data["playlists"] = [playlist_data] - channels_data.append(channel_data) - - results = load_youtube_video_channels(channels_data) - - assert len(results) == len(channels_data) - - for result in results: - assert isinstance(result, VideoChannel) - - assert result.playlists.count() == 1 - - -def test_load_youtube_video_channels_error(mocker): - """Test that an error doesn't fail the entire operation""" - - def pop_channel_id_with_exception(data): - """Pop channel_id off data and raise an exception""" - data.pop("channel_id") - raise ExtractException - - mock_load_channel = mocker.patch( - "learning_resources.etl.loaders.load_video_channel" - ) - mock_load_channel.side_effect = pop_channel_id_with_exception - mock_log = mocker.patch("learning_resources.etl.loaders.log") - channel_id = "abc" - - load_youtube_video_channels([{"channel_id": channel_id}]) - - mock_log.exception.assert_called_once_with( - "Error with extracted video channel: channel_id=%s", channel_id - ) - - -def test_load_youtube_video_channels_unpublish(mock_upsert_tasks): - """Test load_youtube_video_channels when a video/playlist gets unpublished""" +def test_unpublish_removed_youtube_channels(mock_upsert_tasks): + """A channel dropped from the config takes its playlists and videos with it""" channel = VideoChannelFactory.create(etl_source=ETLSource.youtube.name) ovs_channel = VideoChannelFactory.create(etl_source=ETLSource.ovs.name) playlist = VideoPlaylistFactory.create(channel=channel).learning_resource ovs_playlist = VideoPlaylistFactory.create(channel=ovs_channel).learning_resource video = VideoFactory.create().learning_resource - playlist.resources.set( - [video], - through_defaults={ - "relation_type": LearningResourceRelationTypes.PLAYLIST_VIDEOS.value - }, - ) + _add_playlist_videos(playlist, [video]) assert channel.published is True assert video.published is True assert playlist.published is True assert ovs_playlist.published is True - # inputs don't matter here - load_youtube_video_channels([]) + # no channels configured, so the youtube channel is no longer offered + unpublish_removed_youtube_channels([]) video.refresh_from_db() assert video.published is False @@ -3193,6 +3149,7 @@ def test_load_youtube_video_channels_unpublish(mock_upsert_tasks): channel.refresh_from_db() assert channel.published is False + # other ETL sources are left alone ovs_channel.refresh_from_db() assert ovs_channel.published is True ovs_playlist.refresh_from_db() diff --git a/learning_resources/etl/pipelines.py b/learning_resources/etl/pipelines.py index c9498b8bb4..d686582e41 100644 --- a/learning_resources/etl/pipelines.py +++ b/learning_resources/etl/pipelines.py @@ -20,7 +20,6 @@ posthog, sloan, xpro, - youtube, ) from learning_resources.etl.constants import ( CourseLoaderConfig, @@ -162,10 +161,6 @@ def ocw_courses_etl( raise ExtractException(message) -youtube_etl = compose( - loaders.load_youtube_video_channels, youtube.transform, youtube.extract -) - ovs_etl = compose(loaders.load_ovs_playlists, ovs.transform, ovs.extract) posthog_etl = compose( diff --git a/learning_resources/etl/youtube.py b/learning_resources/etl/youtube.py index d98b49d8ea..07e60362da 100644 --- a/learning_resources/etl/youtube.py +++ b/learning_resources/etl/youtube.py @@ -199,7 +199,7 @@ def _extract_playlists( create_videos_channel_setting: bool, ) -> Generator[tuple, None, None]: """ - Extract a list of playlists + Extract the metadata of a list of playlists Args: youtube_client (Resource): Youtube api client @@ -209,7 +209,7 @@ def _extract_playlists( is to create videos from youtube data Returns: - A generator that yields playlist data + A generator that yields (playlist data, create_videos) tuples """ try: while request is not None: @@ -232,11 +232,7 @@ def _extract_playlists( create_videos = create_videos_channel_setting if not playlist_config.get("ignore", False): - yield ( - playlist_data, - extract_playlist_items(youtube_client, playlist_id), - create_videos, - ) + yield (playlist_data, create_videos) request = youtube_client.playlists().list_next(request, response) except StopIteration: @@ -247,7 +243,7 @@ def _extract_playlists( raise ExtractException(msg) from exc -def extract_playlists( +def extract_playlist_metadata( youtube_client: Resource, playlist_configs: list[dict], channel_id: str, @@ -255,15 +251,17 @@ def extract_playlists( create_videos_channel_setting: bool, ) -> Generator[tuple, None, None]: """ - Extract a list of playlists for a channel + Extract the metadata of a channel's playlists, without their videos. + Args: youtube_client (object): Youtube api client playlist_configs (list of dict): list of playlist configurations channel_id (str): youtube's id for the channel create_videos_channel_setting (bool): whether the channel config is to create videos from youtube data + Returns: - A generator that yields playlist data + A generator that yields (playlist data, create_videos) tuples """ playlist_configs_by_id = { @@ -297,59 +295,34 @@ def extract_playlists( ) -def extract_channels( - youtube_client: Resource, channels_config: list[dict] -) -> Generator[tuple, None, None]: +def extract_channel(youtube_client: Resource, channel_id: str) -> dict | None: """ - Extract a list of channels + Extract the raw data for a single channel Args: youtube_client (Resource): Youtube api client - channels_config (list of dict): list of channel configurations + channel_id (str): youtube's id for the channel Returns: - A generator that yields channel data + dict or None: the channel data, or None if youtube has no such channel """ - channel_configs_by_ids = {item["channel_id"]: item for item in channels_config} - channel_ids = set(channel_configs_by_ids.keys()) - - if not channel_ids: - return - try: - request = youtube_client.channels().list( - part="snippet,contentDetails", - id=",".join(channel_ids), - maxResults=YOUTUBE_MAX_RESULTS, + response = ( + youtube_client.channels() + .list( + part="snippet,contentDetails", + id=channel_id, + maxResults=YOUTUBE_MAX_RESULTS, + ) + .execute() ) - - while request is not None: - response = request.execute() - - if response is None: - break - - for channel_data in response["items"]: - channel_id = channel_data["id"] - channel_config = channel_configs_by_ids[channel_id] - offered_by = channel_config.get("offered_by", None) - create_videos = channel_config.get("create_videos", True) - playlist_configs = channel_config.get("playlists", []) - playlists = extract_playlists( - youtube_client, - playlist_configs, - channel_id, - create_videos_channel_setting=create_videos, - ) - yield (offered_by, channel_data, playlists) - - request = youtube_client.channels().list_next(request, response) - except StopIteration: - return except googleapiclient.errors.HttpError as exc: - msg = f"Error fetching channels: channel_ids={channel_ids}" + msg = f"Error fetching channel: channel_id={channel_id}" raise ExtractException(msg) from exc + items = (response or {}).get("items", []) + return items[0] if items else None + def get_captions_for_video(video_resource: LearningResource) -> str: """ @@ -455,27 +428,6 @@ def get_youtube_channel_configs(*, channel_ids: str | None = None) -> list[dict] return channel_configs -def extract(*, channel_ids: str | None = None) -> Generator[tuple, None, None]: - """ - Return video data for all videos in channels' playlists - - Args: - channel_ids (list of str or None): list of channels to extract (all if None) - - Returns: - A generator that yields tuples with offered_by and video data - """ - for setting in ("YOUTUBE_CONFIG_URL", "YOUTUBE_DEVELOPER_KEY"): - if not getattr(settings, setting): - log.error("Missing required setting %s", setting) - return - - youtube_client = get_youtube_client() - channel_configs = get_youtube_channel_configs(channel_ids=channel_ids) - - yield from extract_channels(youtube_client, channel_configs) - - def transform_video(video_data: dict, offered_by_code: str) -> dict: """ Transform raw video data into normalized data structure for single video @@ -550,33 +502,21 @@ def transform_playlist( } -def transform(extracted_channels: iter) -> Generator[dict, None, None]: +def transform_channel(channel_data: dict) -> dict: """ - Transform raw video data into normalized data structure + Transform raw channel data into our normalized data, without its playlists Args: - extracted_channels (iterable of tuple): the youtube channels that were fetched + channel_data (dict): the raw channel data from the youtube api Returns: - generator that yields normalized video data - """ - # NOTE: this generator has nested generators (channels -> playlists -> videos) - # this is by design so that when the loaders run an exception raised in an - # extraction function can signal to the loader code that a partial import occurred - # if you change this it may trigger undefined behavior in the loaders - for offered_by, channel_data, playlists in extracted_channels: - yield { - "channel_id": channel_data["id"], - "title": channel_data["snippet"]["title"], - "published": True, - # intentional generator expression - "playlists": ( - transform_playlist( - playlist, videos, offered_by, create_videos=create_videos - ) - for playlist, videos, create_videos in playlists - ), - } + dict: normalized channel data + """ + return { + "channel_id": channel_data["id"], + "title": channel_data["snippet"]["title"], + "published": True, + } def get_youtube_videos_for_transcripts_job( diff --git a/learning_resources/etl/youtube_test.py b/learning_resources/etl/youtube_test.py index 44d30f53ad..2f2bb4562f 100644 --- a/learning_resources/etl/youtube_test.py +++ b/learning_resources/etl/youtube_test.py @@ -251,36 +251,6 @@ def extracted_and_transformed_values(youtube_api_responses): return extracted, transformed -def _resolve_extracted_channels(channels): - """Resolve the nested generator data""" - return [ - ( - offered_by, - channel_data, - list(map(_resolve_extracted_playlist, playlists)), - ) - for offered_by, channel_data, playlists in channels - ] - - -def _resolve_extracted_playlist(playlist): - """Resolve a playlist and its nested generators""" - playlist_data, videos, create_videos = playlist - return (playlist_data, list(videos), create_videos) - - -@pytest.fixture -def mock_raw_caption_data(): - """Mock data for raw youtube video caption""" - return 'PROFESSOR: So, now we come to\nthe place where arithmetic,modulo n or\nremainder arithmetic,starts to be a little bit\ndifferent and that involvestaking inverses and cancelling.' - - -@pytest.fixture -def mock_parsed_transcript_data(): - """Mock data for parsed video caption""" - return "PROFESSOR: So, now we come to the place where arithmetic,\nmodulo n or remainder arithmetic,\nstarts to be a little bit different and that involves\ntaking inverses and cancelling." - - def test_get_captions_for_video(mocker): """Test fetching caption data for a video when non auto-generated english caption is available""" caption_text = "English: Not Auto-generated" @@ -340,38 +310,6 @@ def test_get_captions_for_video_no_captions(mocker): assert youtube.get_captions_for_video(video) is None -@pytest.mark.usefixtures("mock_youtube_client", "mocked_github_channel_response") -def test_extract(extracted_and_transformed_values): - """Test that extract returns expected responses""" - extracted, _ = extracted_and_transformed_values - results = _resolve_extracted_channels(youtube.extract()) - assert results == extracted - - -@pytest.mark.parametrize( - ("key", "url"), - [ - (None, "https://youtube.test.edu"), - ("key", None), - ], -) -def test_extract_with_unset_keys(settings, key, url): - """Test youtube video ETL extract with no keys set""" - settings.YOUTUBE_DEVELOPER_KEY = key - settings.YOUTUBE_CONFIG_URL = url - - assert _resolve_extracted_channels(youtube.extract()) == [] - - -@pytest.mark.usefixtures("video_settings", "mocked_github_channel_response") -@pytest.mark.parametrize("yaml_parser_response", [None, {}, {"channels": []}]) -def test_extract_with_no_channels(mocker, yaml_parser_response): - """Test youtube video ETL extract with no channels in data""" - mocker.patch("yaml.safe_load", return_value=yaml_parser_response) - - assert _resolve_extracted_channels(youtube.extract()) == [] - - @pytest.mark.django_db @pytest.mark.parametrize( ("error", "raised_exception", "message"), @@ -469,25 +407,37 @@ def test_extract_playlists_create_videos( ) ) assert len(results) == 1 - _, _, create_videos = results[0] + _, create_videos = results[0] assert create_videos is expected -@pytest.mark.django_db -@pytest.mark.parametrize( - ("error", "raised_exception", "message"), - [ - (StopIteration, None, None), - (HttpError, ExtractException, "Error fetching channels: channel_ids="), - ], -) -def test_extract_channels_errors(error, raised_exception, message): - """Test that extract_playlist_items handles errors as expected""" - client = Mock(channels=Mock(side_effect=error(Mock(), b""))) - if raised_exception: - with pytest.raises(raised_exception) as err: - list(youtube.extract_channels(client, [{"channel_id": "channel_id"}])) - assert message in str(err) +@pytest.mark.parametrize("items", [[{"id": "channel_id", "snippet": {}}], []]) +def test_extract_channel(items): + """extract_channel should return the single channel youtube has, if any""" + client = Mock() + client.channels.return_value.list.return_value.execute.return_value = { + "items": items + } + assert youtube.extract_channel(client, "channel_id") == ( + items[0] if items else None + ) + + +def test_extract_channel_error(): + """extract_channel should wrap youtube api errors""" + client = Mock(channels=Mock(side_effect=HttpError(Mock(), b""))) + with pytest.raises(ExtractException) as err: + youtube.extract_channel(client, "channel_id") + assert "Error fetching channel: channel_id=channel_id" in str(err) + + +def test_transform_channel(extracted_and_transformed_values): + """transform_channel should normalize a channel without its playlists""" + extracted, transformed = extracted_and_transformed_values + result = youtube.transform_channel(extracted[0][1]) + assert result == { + key: value for key, value in transformed[0].items() if key != "playlists" + } def test_transform_video(extracted_and_transformed_values): @@ -523,22 +473,6 @@ def test_transform_playlist( } -def test_transform(extracted_and_transformed_values): - """Test youtube transform""" - extracted, transformed = extracted_and_transformed_values - channels = youtube.transform(extracted) - assert [ - { - **channel, - "playlists": [ - {**playlist, "videos": list(playlist["videos"])} - for playlist in channel["playlists"] - ], - } - for channel in channels - ] == transformed - - @pytest.mark.parametrize( ("config", "expected"), [ diff --git a/learning_resources/management/commands/backpopulate_youtube_data.py b/learning_resources/management/commands/backpopulate_youtube_data.py index a72e67f9f2..9d8a1ecad0 100644 --- a/learning_resources/management/commands/backpopulate_youtube_data.py +++ b/learning_resources/management/commands/backpopulate_youtube_data.py @@ -111,8 +111,11 @@ def handle(self, *args, **options): # noqa: ARG002 self.stdout.write(f"Started task {task} to get YouTube video data") self.stdout.write("Waiting on task...") start = now_in_utc() - result = task.get() + channel_count = task.get() total_seconds = (now_in_utc() - start).total_seconds() + # each channel fans out into its own task, and each of those into a + # task per playlist, so the loading continues after this returns self.stdout.write( - f"Fetched {result} YouTube channel in {total_seconds} seconds" + f"Queued {channel_count} YouTube channels in {total_seconds} seconds." + f" Follow the celery logs for the per-channel and per-playlist tasks." ) diff --git a/learning_resources/tasks.py b/learning_resources/tasks.py index 6bd4b0a37e..4510d278d4 100644 --- a/learning_resources/tasks.py +++ b/learning_resources/tasks.py @@ -15,7 +15,7 @@ from django.utils import timezone from learning_resources.constants import LearningResourceType -from learning_resources.etl import ovs, pipelines, youtube +from learning_resources.etl import loaders, ovs, pipelines, youtube from learning_resources.etl.canvas import ( sync_canvas_archive, ) @@ -38,7 +38,7 @@ get_bucket_by_name, get_s3_prefix_for_source, ) -from learning_resources.models import ContentFile, LearningResource +from learning_resources.models import ContentFile, LearningResource, VideoChannel from learning_resources.site_scrapers.utils import scraper_for_site from learning_resources.utils import ( build_program_children_content_bulk, @@ -467,22 +467,135 @@ def get_ocw_data( # noqa: PLR0913 return self.replace(ocw_tasks) -@app.task(acks_late=True) +@app.task(acks_late=True, reject_on_worker_lost=True) +def get_youtube_playlist_data( + channel_id, playlist_data, offered_by_code, *, create_videos +): + """ + Load a single youtube playlist and its videos + + Args: + channel_id (str): youtube's id for the playlist's channel + playlist_data (dict): the raw playlist data from the youtube api + offered_by_code (str): the offered_by code for the playlist + create_videos (bool): whether to create videos from this playlist + or match to existing videos without creating new ones + """ + video_channel = VideoChannel.objects.filter(channel_id=channel_id).first() + if video_channel is None: + # the channel task upserts the channel before fanning out, so this only + # happens if the channel was deleted mid-run + log.error("No VideoChannel for channel_id=%s", channel_id) + return + + youtube_client = youtube.get_youtube_client() + playlist_id = playlist_data["id"] + loaders.load_playlist( + video_channel, + youtube.transform_playlist( + playlist_data, + youtube.extract_playlist_items(youtube_client, playlist_id), + offered_by_code, + create_videos=create_videos, + ), + ) + + +@app.task(acks_late=True, reject_on_worker_lost=True) +def get_youtube_channel_data(channel_config): + """ + Load a single youtube channel and fan its playlists out into their own tasks. + + The channel row is upserted before the playlist tasks are queued so they + can find it, and the channel's full playlist listing is resolved before + anything is unpublished, so a failed extraction can't unpublish a live + playlist. + + Args: + channel_config (dict): the channel's configuration + """ + channel_id = channel_config["channel_id"] + youtube_client = youtube.get_youtube_client() + + channel_data = youtube.extract_channel(youtube_client, channel_id) + if channel_data is None: + log.warning("No youtube data for channel_id=%s", channel_id) + return + + create_videos = channel_config.get("create_videos", True) + playlists = list( + youtube.extract_playlist_metadata( + youtube_client, + channel_config.get("playlists", []), + channel_id, + create_videos_channel_setting=create_videos, + ) + ) + + video_channel = loaders.upsert_video_channel( + youtube.transform_channel(channel_data) + ) + loaders.unpublish_removed_playlists( + video_channel, [playlist_data["id"] for playlist_data, _ in playlists] + ) + + log.info( + "Queueing %d playlists for youtube channel_id=%s", len(playlists), channel_id + ) + for playlist_data, playlist_create_videos in playlists: + get_youtube_playlist_data.delay( + channel_id, + playlist_data, + channel_config.get("offered_by", None), + create_videos=playlist_create_videos, + ) + + +@app.task(acks_late=True, reject_on_worker_lost=True) def get_youtube_data(*, channel_ids=None): """ - Execute the YouTube ETL pipeline + Fan the YouTube ETL out into one task per channel, each of which fans out + into one task per playlist. + + Nothing waits on the fan-out: no worker holds more than a single playlist's + worth of work, so a culled pod costs only the playlist it was loading and + the redelivered message picks it back up. Args: channel_ids (list of str or None): if a list the extraction is limited to those channels Returns: - int: - The number of results that were fetched + int: the number of channels queued """ - results = pipelines.youtube_etl(channel_ids=channel_ids) - clear_views_cache() - return len(list(results)) + missing = [ + setting + for setting in ("YOUTUBE_CONFIG_URL", "YOUTUBE_DEVELOPER_KEY") + if not getattr(settings, setting) + ] + if missing: + log.error("Missing required settings: %s", ", ".join(missing)) + return 0 + + channel_configs = youtube.get_youtube_channel_configs(channel_ids=channel_ids) + if not channel_configs: + # an empty config would unpublish every channel below, so treat it as a + # failure rather than as "youtube offers nothing" + log.error("No youtube channel configs found") + return 0 + + if not channel_ids: + # only a full run knows the complete set of configured channels; a run + # filtered to specific channels must not unpublish the rest + loaders.unpublish_removed_youtube_channels( + [channel_config["channel_id"] for channel_config in channel_configs] + ) + + log.info("Queueing %d youtube channels", len(channel_configs)) + for channel_config in channel_configs: + get_youtube_channel_data.delay(channel_config) + + return len(channel_configs) @app.task diff --git a/learning_resources/tasks_test.py b/learning_resources/tasks_test.py index bfd9d591a0..0a23c93899 100644 --- a/learning_resources/tasks_test.py +++ b/learning_resources/tasks_test.py @@ -14,6 +14,7 @@ from learning_resources.conftest import OCW_TEST_PREFIX, setup_s3, setup_s3_ocw from learning_resources.constants import LearningResourceType, PlatformType from learning_resources.etl.constants import MARKETING_PAGE_FILE_TYPE, ETLSource +from learning_resources.etl.exceptions import ExtractException from learning_resources.factories import ( ContentFileFactory, LearningResourceFactory, @@ -23,7 +24,9 @@ from learning_resources.tasks import ( cleanup_deleted_content_files, get_ocw_data, + get_youtube_channel_data, get_youtube_data, + get_youtube_playlist_data, get_youtube_transcripts, marketing_page_for_resources, scrape_marketing_pages, @@ -83,9 +86,10 @@ def test_cache_is_cleared_after_task_run(mocker, mocked_celery): skip_content_files=True, ) - tasks.get_youtube_data.delay() + # get_youtube_data is absent on purpose: it only queues the fan-out, whose + # writes land long after it returns, so it has nothing to invalidate tasks.get_youtube_transcripts.delay() - assert mocked_clear_views_cache.call_count == 10 + assert mocked_clear_views_cache.call_count == 9 def test_get_mit_edx_data_valid(mocker): @@ -458,12 +462,222 @@ def test_get_ocw_courses(settings, mocker, mocked_celery, timestamp, overwrite): ) -@pytest.mark.parametrize("channel_ids", [["abc", "123"], None]) -def test_get_youtube_data(mocker, settings, channel_ids): - """Verify that the get_youtube_data invokes the YouTube ETL pipeline with expected params""" - mock_pipelines = mocker.patch("learning_resources.tasks.pipelines") - get_youtube_data.delay(channel_ids=channel_ids) - mock_pipelines.youtube_etl.assert_called_once_with(channel_ids=channel_ids) +@pytest.fixture +def youtube_settings(settings): + """Configure youtube ETL settings""" + settings.YOUTUBE_CONFIG_URL = "http://test.youtube/config.yaml" + settings.YOUTUBE_DEVELOPER_KEY = "key" + return settings + + +def _channel_config(channel_id, **kwargs): + """Build a youtube channel config""" + return {"channel_id": channel_id, "offered_by": "ocw", **kwargs} + + +def _playlist_data(playlist_id): + """Build the raw youtube api data for a playlist""" + return { + "id": playlist_id, + "snippet": { + "title": f"Playlist {playlist_id}", + "thumbnails": {"high": {"url": f"http://img/{playlist_id}.jpg"}}, + }, + } + + +@pytest.mark.parametrize("channel_ids", [["channel1"], None]) +def test_get_youtube_data(mocker, youtube_settings, channel_ids): + """get_youtube_data should queue one task per configured channel""" + channel_configs = [_channel_config("channel1"), _channel_config("channel2")] + mock_configs = mocker.patch( + "learning_resources.tasks.youtube.get_youtube_channel_configs", + autospec=True, + return_value=channel_configs, + ) + mock_unpublish = mocker.patch( + "learning_resources.tasks.loaders.unpublish_removed_youtube_channels", + autospec=True, + ) + mock_channel_task = mocker.patch( + "learning_resources.tasks.get_youtube_channel_data", autospec=True + ) + + assert get_youtube_data.delay(channel_ids=channel_ids).get() == 2 + + mock_configs.assert_called_once_with(channel_ids=channel_ids) + assert [ + call.args[0] for call in mock_channel_task.delay.call_args_list + ] == channel_configs + + if channel_ids: + # a run filtered to specific channels doesn't know the full channel set, + # so it must not unpublish the channels it wasn't asked about + mock_unpublish.assert_not_called() + else: + mock_unpublish.assert_called_once_with(["channel1", "channel2"]) + + +def test_get_youtube_data_without_configs_does_not_unpublish(mocker, youtube_settings): + """An empty channel config should be treated as a failure, not as "no channels\"""" + mocker.patch( + "learning_resources.tasks.youtube.get_youtube_channel_configs", + autospec=True, + return_value=[], + ) + mock_unpublish = mocker.patch( + "learning_resources.tasks.loaders.unpublish_removed_youtube_channels", + autospec=True, + ) + mock_channel_task = mocker.patch( + "learning_resources.tasks.get_youtube_channel_data", autospec=True + ) + + assert get_youtube_data.delay().get() == 0 + + mock_unpublish.assert_not_called() + mock_channel_task.delay.assert_not_called() + + +@pytest.mark.parametrize("setting", ["YOUTUBE_CONFIG_URL", "YOUTUBE_DEVELOPER_KEY"]) +def test_get_youtube_data_missing_settings(mocker, youtube_settings, setting): + """A missing youtube setting should stop the run before any extraction""" + setattr(youtube_settings, setting, None) + mock_configs = mocker.patch( + "learning_resources.tasks.youtube.get_youtube_channel_configs", autospec=True + ) + mock_unpublish = mocker.patch( + "learning_resources.tasks.loaders.unpublish_removed_youtube_channels", + autospec=True, + ) + + assert get_youtube_data.delay().get() == 0 + + mock_configs.assert_not_called() + mock_unpublish.assert_not_called() + + +def test_get_youtube_channel_data(mocker, youtube_settings): + """A channel task should load the channel and queue a task per playlist""" + mocker.patch("learning_resources.tasks.youtube.get_youtube_client", autospec=True) + mocker.patch( + "learning_resources.tasks.youtube.extract_channel", + autospec=True, + return_value={"id": "channel1", "snippet": {"title": "Channel 1"}}, + ) + playlists = [_playlist_data("playlist1"), _playlist_data("playlist2")] + mocker.patch( + "learning_resources.tasks.youtube.extract_playlist_metadata", + autospec=True, + return_value=iter([(playlists[0], True), (playlists[1], False)]), + ) + mock_unpublish = mocker.patch( + "learning_resources.tasks.loaders.unpublish_removed_playlists", autospec=True + ) + mock_playlist_task = mocker.patch( + "learning_resources.tasks.get_youtube_playlist_data", autospec=True + ) + + get_youtube_channel_data.delay(_channel_config("channel1")) + + video_channel = models.VideoChannel.objects.get(channel_id="channel1") + assert video_channel.title == "Channel 1" + assert video_channel.published is True + assert video_channel.etl_source == ETLSource.youtube.name + + # the full playlist listing is resolved before anything is unpublished + mock_unpublish.assert_called_once_with(video_channel, ["playlist1", "playlist2"]) + + assert [ + (call.args, call.kwargs) for call in mock_playlist_task.delay.call_args_list + ] == [ + (("channel1", playlists[0], "ocw"), {"create_videos": True}), + (("channel1", playlists[1], "ocw"), {"create_videos": False}), + ] + + +def test_get_youtube_channel_data_missing_from_youtube(mocker, youtube_settings): + """A channel youtube no longer returns should be left alone""" + mocker.patch("learning_resources.tasks.youtube.get_youtube_client", autospec=True) + mocker.patch( + "learning_resources.tasks.youtube.extract_channel", + autospec=True, + return_value=None, + ) + mock_unpublish = mocker.patch( + "learning_resources.tasks.loaders.unpublish_removed_playlists", autospec=True + ) + mock_playlist_task = mocker.patch( + "learning_resources.tasks.get_youtube_playlist_data", autospec=True + ) + + get_youtube_channel_data.delay(_channel_config("channel1")) + + assert models.VideoChannel.objects.count() == 0 + mock_unpublish.assert_not_called() + mock_playlist_task.delay.assert_not_called() + + +def test_get_youtube_channel_data_extract_error_keeps_playlists( + mocker, youtube_settings +): + """A failed playlist listing must not unpublish the channel's playlists""" + mocker.patch("learning_resources.tasks.youtube.get_youtube_client", autospec=True) + mocker.patch( + "learning_resources.tasks.youtube.extract_channel", + autospec=True, + return_value={"id": "channel1", "snippet": {"title": "Channel 1"}}, + ) + mocker.patch( + "learning_resources.tasks.youtube.extract_playlist_metadata", + autospec=True, + side_effect=ExtractException("boom"), + ) + mock_unpublish = mocker.patch( + "learning_resources.tasks.loaders.unpublish_removed_playlists", autospec=True + ) + + with pytest.raises(ExtractException): + get_youtube_channel_data.delay(_channel_config("channel1")) + + mock_unpublish.assert_not_called() + + +def test_get_youtube_playlist_data(mocker, youtube_settings): + """A playlist task should transform and load only its own playlist""" + video_channel = factories.VideoChannelFactory.create(channel_id="channel1") + mocker.patch("learning_resources.tasks.youtube.get_youtube_client", autospec=True) + mock_videos = mocker.patch( + "learning_resources.tasks.youtube.extract_playlist_items", autospec=True + ) + mock_load_playlist = mocker.patch( + "learning_resources.tasks.loaders.load_playlist", autospec=True + ) + + get_youtube_playlist_data.delay( + "channel1", _playlist_data("playlist1"), "ocw", create_videos=True + ) + + mock_videos.assert_called_once_with(ANY, "playlist1") + loaded_channel, playlist_data = mock_load_playlist.call_args.args + assert loaded_channel == video_channel + assert playlist_data["playlist_id"] == "playlist1" + assert playlist_data["create_videos"] is True + assert playlist_data["offered_by"] == {"code": "ocw"} + + +def test_get_youtube_playlist_data_without_channel(mocker, youtube_settings): + """A playlist whose channel vanished mid-run should be skipped, not crash""" + mocker.patch("learning_resources.tasks.youtube.get_youtube_client", autospec=True) + mock_load_playlist = mocker.patch( + "learning_resources.tasks.loaders.load_playlist", autospec=True + ) + + get_youtube_playlist_data.delay( + "channel1", _playlist_data("playlist1"), "ocw", create_videos=True + ) + + mock_load_playlist.assert_not_called() def test_get_youtube_transcripts(mocker):