From bd6b50128a7c94d5df906b3b3f1f46190f7cecfb Mon Sep 17 00:00:00 2001 From: Anastasia Beglova Date: Thu, 6 Aug 2026 12:11:09 -0400 Subject: [PATCH 1/5] first attempt --- learning_resources/etl/loaders.py | 115 +++-- learning_resources/etl/youtube.py | 99 +++- .../commands/backpopulate_youtube_data.py | 57 ++- learning_resources/tasks.py | 134 +++++- learning_resources/tasks_test.py | 440 +++++++++++++++++- 5 files changed, 775 insertions(+), 70 deletions(-) diff --git a/learning_resources/etl/loaders.py b/learning_resources/etl/loaders.py index 0460777b29..046d59c81c 100644 --- a/learning_resources/etl/loaders.py +++ b/learning_resources/etl/loaders.py @@ -1908,9 +1908,31 @@ def load_playlists( return playlists +def upsert_video_channel(video_channel_data: dict) -> VideoChannel: + """ + 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: + VideoChannel: the updated or created video channel + """ + 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 + + def load_video_channel(video_channel_data: dict) -> VideoChannel: """ - Load a single video channel into the database + Load a single video channel, and its playlists, into the database Arg: video_channel_data (dict): @@ -1918,49 +1940,45 @@ def load_video_channel(video_channel_data: dict) -> VideoChannel: Returns: VideoChannel: the updated or created video channel """ - channel_id = video_channel_data.pop("channel_id") playlists_data = video_channel_data.pop("playlists", []) - - video_channel, _ = VideoChannel.objects.select_for_update().update_or_create( - channel_id=channel_id, defaults=video_channel_data - ) + video_channel = upsert_video_channel(video_channel_data) load_playlists(video_channel, playlists_data) return video_channel -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 + """ + playlists_to_unpublish = LearningResource.objects.filter( + video_playlist__channel=video_channel + ).exclude(readable_id__in=playlist_ids) + unpublished_ids = list(playlists_to_unpublish.values_list("id", flat=True)) - Returns: - list of VideoChannel: the loaded video channels + if unpublished_ids: + playlists_to_unpublish.update(published=False) + bulk_resources_unpublished_actions( + unpublished_ids, LearningResourceType.video_playlist.name + ) + + +def unpublish_removed_youtube_channels(channel_ids: list[str]) -> None: """ - 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) + 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) @@ -1996,4 +2014,37 @@ def load_youtube_video_channels(video_channels_data: iter) -> list[VideoChannel] orphaned_video_ids, LearningResourceType.video.name ) + +def load_youtube_video_channels(video_channels_data: iter) -> list[VideoChannel]: + """ + Load a list of video channels + + Args: + video_channels_data (iter of dict): iterable of the video channels data + + 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) + 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) + + unpublish_removed_youtube_channels(channel_ids) + return video_channels diff --git a/learning_resources/etl/youtube.py b/learning_resources/etl/youtube.py index d98b49d8ea..886f6910da 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,6 +295,37 @@ def extract_playlists( ) +def extract_playlists( + youtube_client: Resource, + playlist_configs: list[dict], + channel_id: str, + *, + create_videos_channel_setting: bool, +) -> Generator[tuple, None, None]: + """ + Extract a list of playlists for a channel, with 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 + """ + for playlist_data, create_videos in extract_playlist_metadata( + youtube_client, + playlist_configs, + channel_id, + create_videos_channel_setting=create_videos_channel_setting, + ): + yield ( + playlist_data, + extract_playlist_items(youtube_client, playlist_data["id"]), + create_videos, + ) + + def extract_channels( youtube_client: Resource, channels_config: list[dict] ) -> Generator[tuple, None, None]: @@ -351,6 +380,35 @@ def extract_channels( raise ExtractException(msg) from exc +def extract_channel(youtube_client: Resource, channel_id: str) -> dict | None: + """ + Extract the raw data for a single channel + + Args: + youtube_client (Resource): Youtube api client + channel_id (str): youtube's id for the channel + + Returns: + dict or None: the channel data, or None if youtube has no such channel + """ + try: + response = ( + youtube_client.channels() + .list( + part="snippet,contentDetails", + id=channel_id, + maxResults=YOUTUBE_MAX_RESULTS, + ) + .execute() + ) + except googleapiclient.errors.HttpError as exc: + 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: """ Fetch and return xml captions for a video object @@ -550,6 +608,23 @@ def transform_playlist( } +def transform_channel(channel_data: dict) -> dict: + """ + Transform raw channel data into our normalized data, without its playlists + + Args: + channel_data (dict): the raw channel data from the youtube api + + Returns: + dict: normalized channel data + """ + return { + "channel_id": channel_data["id"], + "title": channel_data["snippet"]["title"], + "published": True, + } + + def transform(extracted_channels: iter) -> Generator[dict, None, None]: """ Transform raw video data into normalized data structure @@ -566,9 +641,7 @@ def transform(extracted_channels: iter) -> Generator[dict, None, None]: # 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, + **transform_channel(channel_data), # intentional generator expression "playlists": ( transform_playlist( diff --git a/learning_resources/management/commands/backpopulate_youtube_data.py b/learning_resources/management/commands/backpopulate_youtube_data.py index a72e67f9f2..bba1b6608e 100644 --- a/learning_resources/management/commands/backpopulate_youtube_data.py +++ b/learning_resources/management/commands/backpopulate_youtube_data.py @@ -3,13 +3,15 @@ from datetime import UTC, datetime from django.core.management import BaseCommand +from django.db.models import Count -from learning_resources.etl.constants import ETLSource +from learning_resources.etl.constants import YOUTUBE_ETL_TASK_NAME, ETLSource from learning_resources.management.commands.mixins import ConfirmDeleteMixin from learning_resources.models import LearningResource, VideoChannel -from learning_resources.tasks import get_youtube_data, get_youtube_transcripts +from learning_resources.tasks import get_youtube_transcripts, start_youtube_etl_job from learning_resources.utils import resource_delete_actions from main.constants import ISOFORMAT +from main.models import TaskJob from main.utils import now_in_utc @@ -59,11 +61,43 @@ def add_arguments(self, parser): action="store_true", help="Overwrite any existing transcript records", ) + mode_group.add_argument( + "--status", + dest="status", + nargs="?", + type=int, + const=0, + default=None, + help=( + "Print the status of a youtube ETL job instead of starting one " + "(defaults to the most recent job)" + ), + ) super().add_arguments(parser) + def print_job_status(self, job_id): + """Print the status of a youtube ETL job""" + jobs = TaskJob.objects.filter(task_name=YOUTUBE_ETL_TASK_NAME) + job = jobs.filter(id=job_id).first() if job_id else jobs.order_by("-id").first() + if not job: + self.stdout.write("No youtube ETL job found") + return + self.stdout.write(f"Youtube ETL job {job.id}: {job.status}") + self.stdout.write(" batches:") + batch_counts = job.batches.values("kind", "status").annotate(count=Count("id")) + kind_counts = {} + for row in batch_counts: + kind_counts.setdefault(row["kind"], {})[row["status"]] = row["count"] + for kind, counts in sorted(kind_counts.items()): + self.stdout.write(f" {kind}: {counts}") + if job.error: + self.stdout.write(f" error: {job.error}") + def handle(self, *args, **options): # noqa: ARG002 """Run Populate youtube videos""" - if options["delete"]: + if options["status"] is not None: + self.print_job_status(options["status"]) + elif options["delete"]: videos_playlists = LearningResource.objects.filter( etl_source=ETLSource.youtube.name ) @@ -106,13 +140,14 @@ def handle(self, *args, **options): # noqa: ARG002 total_seconds = (now_in_utc() - start).total_seconds() self.stdout.write(f"Completed in {total_seconds} seconds") else: - channel_ids = options["channel_ids"] - task = get_youtube_data.delay(channel_ids=channel_ids) - 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() - total_seconds = (now_in_utc() - start).total_seconds() + job = start_youtube_etl_job(channel_ids=options["channel_ids"]) + if job is None: + self.stdout.write( + "A youtube ETL job is already in progress, nothing started" + ) + return + self.stdout.write(f"Started youtube ETL job {job.id}") self.stdout.write( - f"Fetched {result} YouTube channel in {total_seconds} seconds" + f"Check progress with:" + f" ./manage.py backpopulate_youtube_data --status {job.id}" ) diff --git a/learning_resources/tasks.py b/learning_resources/tasks.py index 6bd4b0a37e..651a870258 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,140 @@ 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) + 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. Videos + # orphaned by playlists this run unpublishes are swept up by the next + # full run. + 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) + + # the fan-out writes after this point, but the views cache is short-lived + # and cleared again by the next ETL run, so there's nothing to wait for clear_views_cache() - return len(list(results)) + return len(channel_configs) @app.task diff --git a/learning_resources/tasks_test.py b/learning_resources/tasks_test.py index bfd9d591a0..ac9aa6eac1 100644 --- a/learning_resources/tasks_test.py +++ b/learning_resources/tasks_test.py @@ -13,7 +13,13 @@ from learning_resources import factories, models, tasks 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.constants import ( + MARKETING_PAGE_FILE_TYPE, + YOUTUBE_ETL_TASK_NAME, + ETLSource, + YoutubeBatchKind, +) +from learning_resources.etl.exceptions import ExtractException from learning_resources.factories import ( ContentFileFactory, LearningResourceFactory, @@ -22,15 +28,20 @@ from learning_resources.models import ContentFile, LearningResource from learning_resources.tasks import ( cleanup_deleted_content_files, + finish_youtube_etl, get_ocw_data, get_youtube_data, get_youtube_transcripts, marketing_page_for_resources, + run_youtube_batch, scrape_marketing_pages, + start_youtube_etl, sync_canvas_courses, update_next_start_date_and_prices, update_ocw_learning_material_resources, ) +from main.factories import TaskBatchFactory, TaskJobFactory +from main.models import TaskBatch, TaskJob from main.utils import now_in_utc pytestmark = pytest.mark.django_db @@ -458,12 +469,429 @@ def test_get_ocw_courses(settings, mocker, mocked_celery, timestamp, overwrite): ) +@pytest.fixture +def youtube_settings(settings): + """Settings with the youtube ETL configured""" + 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", [["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) +def test_get_youtube_data(mocker, channel_ids): + """get_youtube_data should create a job and enqueue its start task""" + mock_start = mocker.patch( + "learning_resources.tasks.start_youtube_etl", autospec=True + ) + + job_id = get_youtube_data.delay(channel_ids=channel_ids).get() + + job = TaskJob.objects.get(id=job_id) + assert job.task_name == YOUTUBE_ETL_TASK_NAME + assert job.status == TaskJob.Status.QUEUED + assert job.params["channel_ids"] == channel_ids + mock_start.delay.assert_called_once_with(job_id) + + +@pytest.mark.parametrize("status", TaskJob.ACTIVE_STATUSES) +def test_get_youtube_data_skips_active_job(mocker, status): + """get_youtube_data should not start a second job while one is in progress""" + TaskJobFactory.create(task_name=YOUTUBE_ETL_TASK_NAME, status=status) + mock_start = mocker.patch( + "learning_resources.tasks.start_youtube_etl", autospec=True + ) + + assert get_youtube_data.delay().get() is None + assert TaskJob.objects.filter(task_name=YOUTUBE_ETL_TASK_NAME).count() == 1 + mock_start.delay.assert_not_called() + + +def test_start_youtube_etl(mocker, youtube_settings): + """start_youtube_etl should create a batch per configured channel""" + channel_configs = [_channel_config("channel1"), _channel_config("channel2")] + mocker.patch( + "learning_resources.tasks.youtube.get_youtube_channel_configs", + autospec=True, + return_value=channel_configs, + ) + mock_run_batch = mocker.patch( + "learning_resources.tasks.run_youtube_batch", autospec=True + ) + + job = TaskJobFactory.create( + task_name=YOUTUBE_ETL_TASK_NAME, params={"channel_ids": None} + ) + start_youtube_etl.delay(job.id) + + job.refresh_from_db() + assert job.status == TaskJob.Status.RUNNING + assert job.params["channel_ids"] == ["channel1", "channel2"] + + batches = job.batches.order_by("batch_key") + assert [batch.batch_key for batch in batches] == [ + "channel:channel1", + "channel:channel2", + ] + assert {batch.kind for batch in batches} == {YoutubeBatchKind.channel.value} + assert [batch.params["channel_config"] for batch in batches] == channel_configs + assert sorted( + call.args[0] for call in mock_run_batch.delay.call_args_list + ) == sorted(batch.id for batch in batches) + + +def test_start_youtube_etl_requeues_pending_batches(mocker, youtube_settings): + """A redelivered start task should re-enqueue queued batches without rebuilding them""" + mock_configs = mocker.patch( + "learning_resources.tasks.youtube.get_youtube_channel_configs", autospec=True + ) + mock_run_batch = mocker.patch( + "learning_resources.tasks.run_youtube_batch", autospec=True + ) + + job = TaskJobFactory.create( + task_name=YOUTUBE_ETL_TASK_NAME, + status=TaskJob.Status.RUNNING, + params={"channel_ids": ["channel1"]}, + ) + queued = TaskBatchFactory.create(job=job, status=TaskBatch.Status.QUEUED) + TaskBatchFactory.create(job=job, status=TaskBatch.Status.SUCCEEDED) + + start_youtube_etl.delay(job.id) + + mock_configs.assert_not_called() + mock_run_batch.delay.assert_called_once_with(queued.id) + + +def test_start_youtube_etl_without_configs_does_not_unpublish(mocker, youtube_settings): + """An empty channel config should fail the job rather than unpublish everything""" + mocker.patch( + "learning_resources.tasks.youtube.get_youtube_channel_configs", + autospec=True, + return_value=[], + ) + mock_finish = mocker.patch( + "learning_resources.tasks.finish_youtube_etl", autospec=True + ) + + job = TaskJobFactory.create( + task_name=YOUTUBE_ETL_TASK_NAME, params={"channel_ids": None} + ) + start_youtube_etl.delay(job.id) + + job.refresh_from_db() + assert job.status == TaskJob.Status.FAILED + assert job.error == "No youtube channel configs found" + assert job.batches.count() == 0 + mock_finish.delay.assert_not_called() + + +@pytest.mark.parametrize("setting", ["YOUTUBE_CONFIG_URL", "YOUTUBE_DEVELOPER_KEY"]) +def test_start_youtube_etl_missing_settings(mocker, youtube_settings, setting): + """A missing youtube setting should fail the job before any extraction""" + setattr(youtube_settings, setting, None) + mock_configs = mocker.patch( + "learning_resources.tasks.youtube.get_youtube_channel_configs", autospec=True + ) + + job = TaskJobFactory.create( + task_name=YOUTUBE_ETL_TASK_NAME, params={"channel_ids": None} + ) + start_youtube_etl.delay(job.id) + + job.refresh_from_db() + assert job.status == TaskJob.Status.FAILED + assert setting in job.error + mock_configs.assert_not_called() + + +def test_run_youtube_batch_channel(mocker, youtube_settings): + """A channel batch should load the channel and fan its playlists out into batches""" + 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, + return_value=iter( + [(_playlist_data("playlist1"), True), (_playlist_data("playlist2"), False)] + ), + ) + mock_unpublish = mocker.patch( + "learning_resources.tasks.loaders.unpublish_removed_playlists", autospec=True + ) + mock_run_batch = mocker.patch( + "learning_resources.tasks.run_youtube_batch", autospec=True + ) + + job = TaskJobFactory.create( + task_name=YOUTUBE_ETL_TASK_NAME, + status=TaskJob.Status.RUNNING, + params={"channel_ids": ["channel1"]}, + ) + batch = TaskBatchFactory.create( + job=job, + kind=YoutubeBatchKind.channel.value, + batch_key="channel:channel1", + params={"channel_config": _channel_config("channel1")}, + ) + + run_youtube_batch(batch.id) + + batch.refresh_from_db() + assert batch.status == TaskBatch.Status.SUCCEEDED + + 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 + + mock_unpublish.assert_called_once_with(video_channel, ["playlist1", "playlist2"]) + + playlist_batches = job.batches.filter( + kind=YoutubeBatchKind.playlist.value + ).order_by("batch_key") + assert [playlist_batch.batch_key for playlist_batch in playlist_batches] == [ + "playlist:playlist1", + "playlist:playlist2", + ] + assert [ + playlist_batch.params["create_videos"] for playlist_batch in playlist_batches + ] == [True, False] + assert all( + playlist_batch.params["channel_id"] == "channel1" + and playlist_batch.params["offered_by"] == "ocw" + for playlist_batch in playlist_batches + ) + assert sorted( + call.args[0] for call in mock_run_batch.delay.call_args_list + ) == sorted(playlist_batch.id for playlist_batch in playlist_batches) + + +def test_run_youtube_batch_channel_missing_from_youtube(mocker, youtube_settings): + """A channel youtube no longer returns should succeed without creating batches""" + 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 + ) + + job = TaskJobFactory.create( + task_name=YOUTUBE_ETL_TASK_NAME, + status=TaskJob.Status.RUNNING, + params={"channel_ids": ["channel1"]}, + ) + batch = TaskBatchFactory.create( + job=job, + kind=YoutubeBatchKind.channel.value, + params={"channel_config": _channel_config("channel1")}, + ) + + run_youtube_batch(batch.id) + + batch.refresh_from_db() + assert batch.status == TaskBatch.Status.SUCCEEDED + assert models.VideoChannel.objects.count() == 0 + assert job.batches.filter(kind=YoutubeBatchKind.playlist.value).count() == 0 + mock_unpublish.assert_not_called() + + +def test_run_youtube_batch_playlist(mocker, youtube_settings): + """A playlist batch should transform and load just 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 + ) + + job = TaskJobFactory.create( + task_name=YOUTUBE_ETL_TASK_NAME, + status=TaskJob.Status.RUNNING, + params={"channel_ids": ["channel1"]}, + ) + batch = TaskBatchFactory.create( + job=job, + kind=YoutubeBatchKind.playlist.value, + params={ + "channel_id": "channel1", + "playlist_data": _playlist_data("playlist1"), + "offered_by": "ocw", + "create_videos": True, + }, + ) + + run_youtube_batch(batch.id) + + batch.refresh_from_db() + assert batch.status == TaskBatch.Status.SUCCEEDED + 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 + + +def test_run_youtube_batch_records_failure(mocker, youtube_settings): + """A batch that raises should be marked failed rather than left running""" + mocker.patch("learning_resources.tasks.youtube.get_youtube_client", autospec=True) + mocker.patch( + "learning_resources.tasks.youtube.extract_channel", + autospec=True, + side_effect=ExtractException("boom"), + ) + mock_finish = mocker.patch( + "learning_resources.tasks.finish_youtube_etl", autospec=True + ) + + job = TaskJobFactory.create( + task_name=YOUTUBE_ETL_TASK_NAME, + status=TaskJob.Status.RUNNING, + params={"channel_ids": ["channel1"]}, + ) + batch = TaskBatchFactory.create( + job=job, + kind=YoutubeBatchKind.channel.value, + params={"channel_config": _channel_config("channel1")}, + ) + + run_youtube_batch(batch.id) + + batch.refresh_from_db() + assert batch.status == TaskBatch.Status.FAILED + assert "boom" in batch.error + # the failure still completes the job so it can't hang + job.refresh_from_db() + assert job.status == TaskJob.Status.FINISHING + mock_finish.delay.assert_called_once_with(job.id) + + +def test_run_youtube_batch_skips_terminal_batch(mocker): + """A redelivered batch that already finished should not be re-run""" + mock_execute = mocker.patch( + "learning_resources.tasks._execute_youtube_batch", autospec=True + ) + mock_finish = mocker.patch( + "learning_resources.tasks.finish_youtube_etl", autospec=True + ) + + job = TaskJobFactory.create( + task_name=YOUTUBE_ETL_TASK_NAME, + status=TaskJob.Status.RUNNING, + params={"channel_ids": ["channel1"]}, + ) + batch = TaskBatchFactory.create(job=job, status=TaskBatch.Status.SUCCEEDED) + + run_youtube_batch(batch.id) + + mock_execute.assert_not_called() + # the redelivery still nudges the job to completion + job.refresh_from_db() + assert job.status == TaskJob.Status.FINISHING + mock_finish.delay.assert_called_once_with(job.id) + + +def test_run_youtube_batch_finishes_job_only_when_all_batches_done(mocker): + """The finish step should wait for every batch of the job""" + mocker.patch("learning_resources.tasks._execute_youtube_batch", autospec=True) + mock_finish = mocker.patch( + "learning_resources.tasks.finish_youtube_etl", autospec=True + ) + + job = TaskJobFactory.create( + task_name=YOUTUBE_ETL_TASK_NAME, + status=TaskJob.Status.RUNNING, + params={"channel_ids": ["channel1"]}, + ) + batches = TaskBatchFactory.create_batch(2, job=job) + + run_youtube_batch(batches[0].id) + mock_finish.delay.assert_not_called() + + run_youtube_batch(batches[1].id) + mock_finish.delay.assert_called_once_with(job.id) + + +@pytest.mark.parametrize("has_failures", [True, False]) +def test_finish_youtube_etl(mocker, has_failures): + """finish_youtube_etl should unpublish removed channels and close out the job""" + mock_unpublish = mocker.patch( + "learning_resources.tasks.loaders.unpublish_removed_youtube_channels", + autospec=True, + ) + mock_clear_cache = mocker.patch( + "learning_resources.tasks.clear_views_cache", autospec=True + ) + + job = TaskJobFactory.create( + task_name=YOUTUBE_ETL_TASK_NAME, + status=TaskJob.Status.FINISHING, + params={"channel_ids": ["channel1", "channel2"]}, + ) + TaskBatchFactory.create(job=job, status=TaskBatch.Status.SUCCEEDED) + if has_failures: + TaskBatchFactory.create( + job=job, + batch_key="channel:channel2", + status=TaskBatch.Status.FAILED, + error="boom", + ) + + finish_youtube_etl.delay(job.id) + + # cleanup runs either way; a failed batch leaves its resources as they were + mock_unpublish.assert_called_once_with(["channel1", "channel2"]) + mock_clear_cache.assert_called_once() + + job.refresh_from_db() + if has_failures: + assert job.status == TaskJob.Status.FAILED + assert "channel:channel2: boom" in job.error + else: + assert job.status == TaskJob.Status.SUCCEEDED + assert job.error == "" + + +def test_finish_youtube_etl_skips_unclaimed_job(mocker): + """finish_youtube_etl should do nothing for a job it hasn't claimed""" + mock_unpublish = mocker.patch( + "learning_resources.tasks.loaders.unpublish_removed_youtube_channels", + autospec=True, + ) + + job = TaskJobFactory.create( + task_name=YOUTUBE_ETL_TASK_NAME, + status=TaskJob.Status.SUCCEEDED, + params={"channel_ids": ["channel1"]}, + ) + finish_youtube_etl.delay(job.id) + + mock_unpublish.assert_not_called() def test_get_youtube_transcripts(mocker): From 205a85638631e491e6c49d0d742107a057738d61 Mon Sep 17 00:00:00 2001 From: Anastasia Beglova Date: Thu, 6 Aug 2026 15:46:07 -0400 Subject: [PATCH 2/5] commit --- learning_resources/etl/loaders_test.py | 41 ++ learning_resources/etl/youtube_test.py | 31 +- .../commands/backpopulate_youtube_data.py | 60 +-- learning_resources/tasks_test.py | 427 +++++------------- 4 files changed, 199 insertions(+), 360 deletions(-) diff --git a/learning_resources/etl/loaders_test.py b/learning_resources/etl/loaders_test.py index fa8c962b4a..8bdfc91211 100644 --- a/learning_resources/etl/loaders_test.py +++ b/learning_resources/etl/loaders_test.py @@ -63,6 +63,7 @@ load_videos, load_videos_from_content_files, load_youtube_video_channels, + unpublish_removed_playlists, ) from learning_resources.etl.mitxonline import transform_programs from learning_resources.etl.utils import get_s3_prefix_for_source @@ -2867,6 +2868,46 @@ def test_load_playlists_unpublish(mocker): assert actual_unpublished_ids == expected_unpublished_ids +def test_unpublish_removed_playlists(mocker): + """Playlists no longer listed under the channel should be unpublished""" + mocker.patch("learning_resources_search.tasks.bulk_deindex_learning_resources.si") + mock_bulk_unpublish = mocker.patch( + "learning_resources.etl.loaders.bulk_resources_unpublished_actions", + ) + channel = VideoChannelFactory.create() + kept, removed = VideoPlaylistFactory.create_batch(2, channel=channel) + other_channel_playlist = VideoPlaylistFactory.create() + + 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 + + mock_bulk_unpublish.assert_called_once_with( + [removed.learning_resource.id], LearningResourceType.video_playlist.name + ) + + +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 = VideoPlaylistFactory.create_batch(2, channel=channel) + + unpublish_removed_playlists( + channel, [playlist.learning_resource.readable_id for playlist in playlists] + ) + + mock_bulk_unpublish.assert_not_called() + + @pytest.mark.parametrize("playlist_exists", [True, False]) def test_load_ovs_playlist(mocker, playlist_exists, mock_get_similar_topics_qdrant): """Test load_ovs_playlist creates/updates a playlist and its videos""" diff --git a/learning_resources/etl/youtube_test.py b/learning_resources/etl/youtube_test.py index 44d30f53ad..046f075dd4 100644 --- a/learning_resources/etl/youtube_test.py +++ b/learning_resources/etl/youtube_test.py @@ -469,7 +469,7 @@ def test_extract_playlists_create_videos( ) ) assert len(results) == 1 - _, _, create_videos = results[0] + _, create_videos = results[0] assert create_videos is expected @@ -490,6 +490,35 @@ def test_extract_channels_errors(error, raised_exception, message): 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): """Test youtube transform for a video""" extracted, transformed = extracted_and_transformed_values diff --git a/learning_resources/management/commands/backpopulate_youtube_data.py b/learning_resources/management/commands/backpopulate_youtube_data.py index bba1b6608e..9d8a1ecad0 100644 --- a/learning_resources/management/commands/backpopulate_youtube_data.py +++ b/learning_resources/management/commands/backpopulate_youtube_data.py @@ -3,15 +3,13 @@ from datetime import UTC, datetime from django.core.management import BaseCommand -from django.db.models import Count -from learning_resources.etl.constants import YOUTUBE_ETL_TASK_NAME, ETLSource +from learning_resources.etl.constants import ETLSource from learning_resources.management.commands.mixins import ConfirmDeleteMixin from learning_resources.models import LearningResource, VideoChannel -from learning_resources.tasks import get_youtube_transcripts, start_youtube_etl_job +from learning_resources.tasks import get_youtube_data, get_youtube_transcripts from learning_resources.utils import resource_delete_actions from main.constants import ISOFORMAT -from main.models import TaskJob from main.utils import now_in_utc @@ -61,43 +59,11 @@ def add_arguments(self, parser): action="store_true", help="Overwrite any existing transcript records", ) - mode_group.add_argument( - "--status", - dest="status", - nargs="?", - type=int, - const=0, - default=None, - help=( - "Print the status of a youtube ETL job instead of starting one " - "(defaults to the most recent job)" - ), - ) super().add_arguments(parser) - def print_job_status(self, job_id): - """Print the status of a youtube ETL job""" - jobs = TaskJob.objects.filter(task_name=YOUTUBE_ETL_TASK_NAME) - job = jobs.filter(id=job_id).first() if job_id else jobs.order_by("-id").first() - if not job: - self.stdout.write("No youtube ETL job found") - return - self.stdout.write(f"Youtube ETL job {job.id}: {job.status}") - self.stdout.write(" batches:") - batch_counts = job.batches.values("kind", "status").annotate(count=Count("id")) - kind_counts = {} - for row in batch_counts: - kind_counts.setdefault(row["kind"], {})[row["status"]] = row["count"] - for kind, counts in sorted(kind_counts.items()): - self.stdout.write(f" {kind}: {counts}") - if job.error: - self.stdout.write(f" error: {job.error}") - def handle(self, *args, **options): # noqa: ARG002 """Run Populate youtube videos""" - if options["status"] is not None: - self.print_job_status(options["status"]) - elif options["delete"]: + if options["delete"]: videos_playlists = LearningResource.objects.filter( etl_source=ETLSource.youtube.name ) @@ -140,14 +106,16 @@ def handle(self, *args, **options): # noqa: ARG002 total_seconds = (now_in_utc() - start).total_seconds() self.stdout.write(f"Completed in {total_seconds} seconds") else: - job = start_youtube_etl_job(channel_ids=options["channel_ids"]) - if job is None: - self.stdout.write( - "A youtube ETL job is already in progress, nothing started" - ) - return - self.stdout.write(f"Started youtube ETL job {job.id}") + channel_ids = options["channel_ids"] + task = get_youtube_data.delay(channel_ids=channel_ids) + self.stdout.write(f"Started task {task} to get YouTube video data") + self.stdout.write("Waiting on task...") + start = now_in_utc() + 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"Check progress with:" - f" ./manage.py backpopulate_youtube_data --status {job.id}" + 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_test.py b/learning_resources/tasks_test.py index ac9aa6eac1..3e4f37d9eb 100644 --- a/learning_resources/tasks_test.py +++ b/learning_resources/tasks_test.py @@ -13,12 +13,7 @@ from learning_resources import factories, models, tasks 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, - YOUTUBE_ETL_TASK_NAME, - ETLSource, - YoutubeBatchKind, -) +from learning_resources.etl.constants import MARKETING_PAGE_FILE_TYPE, ETLSource from learning_resources.etl.exceptions import ExtractException from learning_resources.factories import ( ContentFileFactory, @@ -28,20 +23,17 @@ from learning_resources.models import ContentFile, LearningResource from learning_resources.tasks import ( cleanup_deleted_content_files, - finish_youtube_etl, get_ocw_data, + get_youtube_channel_data, get_youtube_data, + get_youtube_playlist_data, get_youtube_transcripts, marketing_page_for_resources, - run_youtube_batch, scrape_marketing_pages, - start_youtube_etl, sync_canvas_courses, update_next_start_date_and_prices, update_ocw_learning_material_resources, ) -from main.factories import TaskBatchFactory, TaskJobFactory -from main.models import TaskBatch, TaskJob from main.utils import now_in_utc pytestmark = pytest.mark.django_db @@ -72,11 +64,23 @@ def mock_blocklist(mocker): ) -def test_cache_is_cleared_after_task_run(mocker, mocked_celery): +def test_cache_is_cleared_after_task_run(mocker, mocked_celery, settings): """Test that the search cache is cleared out after every task run""" mocker.patch("learning_resources.tasks.ocw_courses_etl", autospec=True) mocker.patch("learning_resources.tasks.get_content_tasks", autospec=True) mocker.patch("learning_resources.tasks.pipelines") + settings.YOUTUBE_CONFIG_URL = "http://test.youtube/config.yaml" + settings.YOUTUBE_DEVELOPER_KEY = "key" + mocker.patch( + "learning_resources.tasks.youtube.get_youtube_channel_configs", + autospec=True, + return_value=[{"channel_id": "channel1"}], + ) + mocker.patch( + "learning_resources.tasks.loaders.unpublish_removed_youtube_channels", + autospec=True, + ) + mocker.patch("learning_resources.tasks.get_youtube_channel_data", autospec=True) mocked_clear_views_cache = mocker.patch( "learning_resources.tasks.clear_views_cache" ) @@ -471,7 +475,7 @@ def test_get_ocw_courses(settings, mocker, mocked_celery, timestamp, overwrite): @pytest.fixture def youtube_settings(settings): - """Settings with the youtube ETL configured""" + """Configure youtube ETL settings""" settings.YOUTUBE_CONFIG_URL = "http://test.youtube/config.yaml" settings.YOUTUBE_DEVELOPER_KEY = "key" return settings @@ -493,201 +497,122 @@ def _playlist_data(playlist_id): } -@pytest.mark.parametrize("channel_ids", [["abc", "123"], None]) -def test_get_youtube_data(mocker, channel_ids): - """get_youtube_data should create a job and enqueue its start task""" - mock_start = mocker.patch( - "learning_resources.tasks.start_youtube_etl", autospec=True - ) - - job_id = get_youtube_data.delay(channel_ids=channel_ids).get() - - job = TaskJob.objects.get(id=job_id) - assert job.task_name == YOUTUBE_ETL_TASK_NAME - assert job.status == TaskJob.Status.QUEUED - assert job.params["channel_ids"] == channel_ids - mock_start.delay.assert_called_once_with(job_id) - - -@pytest.mark.parametrize("status", TaskJob.ACTIVE_STATUSES) -def test_get_youtube_data_skips_active_job(mocker, status): - """get_youtube_data should not start a second job while one is in progress""" - TaskJobFactory.create(task_name=YOUTUBE_ETL_TASK_NAME, status=status) - mock_start = mocker.patch( - "learning_resources.tasks.start_youtube_etl", autospec=True - ) - - assert get_youtube_data.delay().get() is None - assert TaskJob.objects.filter(task_name=YOUTUBE_ETL_TASK_NAME).count() == 1 - mock_start.delay.assert_not_called() - - -def test_start_youtube_etl(mocker, youtube_settings): - """start_youtube_etl should create a batch per configured channel""" +@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")] - mocker.patch( + mock_configs = mocker.patch( "learning_resources.tasks.youtube.get_youtube_channel_configs", autospec=True, return_value=channel_configs, ) - mock_run_batch = mocker.patch( - "learning_resources.tasks.run_youtube_batch", autospec=True + mock_unpublish = mocker.patch( + "learning_resources.tasks.loaders.unpublish_removed_youtube_channels", + autospec=True, ) - - job = TaskJobFactory.create( - task_name=YOUTUBE_ETL_TASK_NAME, params={"channel_ids": None} + mock_channel_task = mocker.patch( + "learning_resources.tasks.get_youtube_channel_data", autospec=True ) - start_youtube_etl.delay(job.id) - - job.refresh_from_db() - assert job.status == TaskJob.Status.RUNNING - assert job.params["channel_ids"] == ["channel1", "channel2"] - - batches = job.batches.order_by("batch_key") - assert [batch.batch_key for batch in batches] == [ - "channel:channel1", - "channel:channel2", - ] - assert {batch.kind for batch in batches} == {YoutubeBatchKind.channel.value} - assert [batch.params["channel_config"] for batch in batches] == channel_configs - assert sorted( - call.args[0] for call in mock_run_batch.delay.call_args_list - ) == sorted(batch.id for batch in batches) - - -def test_start_youtube_etl_requeues_pending_batches(mocker, youtube_settings): - """A redelivered start task should re-enqueue queued batches without rebuilding them""" - mock_configs = mocker.patch( - "learning_resources.tasks.youtube.get_youtube_channel_configs", autospec=True - ) - mock_run_batch = mocker.patch( - "learning_resources.tasks.run_youtube_batch", autospec=True + mock_clear_cache = mocker.patch( + "learning_resources.tasks.clear_views_cache", autospec=True ) - job = TaskJobFactory.create( - task_name=YOUTUBE_ETL_TASK_NAME, - status=TaskJob.Status.RUNNING, - params={"channel_ids": ["channel1"]}, - ) - queued = TaskBatchFactory.create(job=job, status=TaskBatch.Status.QUEUED) - TaskBatchFactory.create(job=job, status=TaskBatch.Status.SUCCEEDED) + assert get_youtube_data.delay(channel_ids=channel_ids).get() == 2 - start_youtube_etl.delay(job.id) + 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 + mock_clear_cache.assert_called_once() - mock_configs.assert_not_called() - mock_run_batch.delay.assert_called_once_with(queued.id) + 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_start_youtube_etl_without_configs_does_not_unpublish(mocker, youtube_settings): - """An empty channel config should fail the job rather than unpublish everything""" +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_finish = mocker.patch( - "learning_resources.tasks.finish_youtube_etl", autospec=True + mock_unpublish = mocker.patch( + "learning_resources.tasks.loaders.unpublish_removed_youtube_channels", + autospec=True, ) - - job = TaskJobFactory.create( - task_name=YOUTUBE_ETL_TASK_NAME, params={"channel_ids": None} + mock_channel_task = mocker.patch( + "learning_resources.tasks.get_youtube_channel_data", autospec=True ) - start_youtube_etl.delay(job.id) - job.refresh_from_db() - assert job.status == TaskJob.Status.FAILED - assert job.error == "No youtube channel configs found" - assert job.batches.count() == 0 - mock_finish.delay.assert_not_called() + 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_start_youtube_etl_missing_settings(mocker, youtube_settings, setting): - """A missing youtube setting should fail the job before any extraction""" +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 ) - - job = TaskJobFactory.create( - task_name=YOUTUBE_ETL_TASK_NAME, params={"channel_ids": None} + mock_unpublish = mocker.patch( + "learning_resources.tasks.loaders.unpublish_removed_youtube_channels", + autospec=True, ) - start_youtube_etl.delay(job.id) - job.refresh_from_db() - assert job.status == TaskJob.Status.FAILED - assert setting in job.error + assert get_youtube_data.delay().get() == 0 + mock_configs.assert_not_called() + mock_unpublish.assert_not_called() -def test_run_youtube_batch_channel(mocker, youtube_settings): - """A channel batch should load the channel and fan its playlists out into batches""" +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( - [(_playlist_data("playlist1"), True), (_playlist_data("playlist2"), False)] - ), + return_value=iter([(playlists[0], True), (playlists[1], False)]), ) mock_unpublish = mocker.patch( "learning_resources.tasks.loaders.unpublish_removed_playlists", autospec=True ) - mock_run_batch = mocker.patch( - "learning_resources.tasks.run_youtube_batch", autospec=True - ) - - job = TaskJobFactory.create( - task_name=YOUTUBE_ETL_TASK_NAME, - status=TaskJob.Status.RUNNING, - params={"channel_ids": ["channel1"]}, - ) - batch = TaskBatchFactory.create( - job=job, - kind=YoutubeBatchKind.channel.value, - batch_key="channel:channel1", - params={"channel_config": _channel_config("channel1")}, + mock_playlist_task = mocker.patch( + "learning_resources.tasks.get_youtube_playlist_data", autospec=True ) - run_youtube_batch(batch.id) - - batch.refresh_from_db() - assert batch.status == TaskBatch.Status.SUCCEEDED + 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"]) - playlist_batches = job.batches.filter( - kind=YoutubeBatchKind.playlist.value - ).order_by("batch_key") - assert [playlist_batch.batch_key for playlist_batch in playlist_batches] == [ - "playlist:playlist1", - "playlist:playlist2", - ] assert [ - playlist_batch.params["create_videos"] for playlist_batch in playlist_batches - ] == [True, False] - assert all( - playlist_batch.params["channel_id"] == "channel1" - and playlist_batch.params["offered_by"] == "ocw" - for playlist_batch in playlist_batches - ) - assert sorted( - call.args[0] for call in mock_run_batch.delay.call_args_list - ) == sorted(playlist_batch.id for playlist_batch in playlist_batches) + (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_run_youtube_batch_channel_missing_from_youtube(mocker, youtube_settings): - """A channel youtube no longer returns should succeed without creating batches""" +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", @@ -697,29 +622,44 @@ def test_run_youtube_batch_channel_missing_from_youtube(mocker, youtube_settings 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() - job = TaskJobFactory.create( - task_name=YOUTUBE_ETL_TASK_NAME, - status=TaskJob.Status.RUNNING, - params={"channel_ids": ["channel1"]}, + +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"}}, ) - batch = TaskBatchFactory.create( - job=job, - kind=YoutubeBatchKind.channel.value, - params={"channel_config": _channel_config("channel1")}, + 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 ) - run_youtube_batch(batch.id) + with pytest.raises(ExtractException): + get_youtube_channel_data.delay(_channel_config("channel1")) - batch.refresh_from_db() - assert batch.status == TaskBatch.Status.SUCCEEDED - assert models.VideoChannel.objects.count() == 0 - assert job.batches.filter(kind=YoutubeBatchKind.playlist.value).count() == 0 mock_unpublish.assert_not_called() -def test_run_youtube_batch_playlist(mocker, youtube_settings): - """A playlist batch should transform and load just its own playlist""" +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( @@ -729,169 +669,30 @@ def test_run_youtube_batch_playlist(mocker, youtube_settings): "learning_resources.tasks.loaders.load_playlist", autospec=True ) - job = TaskJobFactory.create( - task_name=YOUTUBE_ETL_TASK_NAME, - status=TaskJob.Status.RUNNING, - params={"channel_ids": ["channel1"]}, - ) - batch = TaskBatchFactory.create( - job=job, - kind=YoutubeBatchKind.playlist.value, - params={ - "channel_id": "channel1", - "playlist_data": _playlist_data("playlist1"), - "offered_by": "ocw", - "create_videos": True, - }, + get_youtube_playlist_data.delay( + "channel1", _playlist_data("playlist1"), "ocw", create_videos=True ) - run_youtube_batch(batch.id) - - batch.refresh_from_db() - assert batch.status == TaskBatch.Status.SUCCEEDED 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_run_youtube_batch_records_failure(mocker, youtube_settings): - """A batch that raises should be marked failed rather than left running""" +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) - mocker.patch( - "learning_resources.tasks.youtube.extract_channel", - autospec=True, - side_effect=ExtractException("boom"), - ) - mock_finish = mocker.patch( - "learning_resources.tasks.finish_youtube_etl", autospec=True - ) - - job = TaskJobFactory.create( - task_name=YOUTUBE_ETL_TASK_NAME, - status=TaskJob.Status.RUNNING, - params={"channel_ids": ["channel1"]}, - ) - batch = TaskBatchFactory.create( - job=job, - kind=YoutubeBatchKind.channel.value, - params={"channel_config": _channel_config("channel1")}, - ) - - run_youtube_batch(batch.id) - - batch.refresh_from_db() - assert batch.status == TaskBatch.Status.FAILED - assert "boom" in batch.error - # the failure still completes the job so it can't hang - job.refresh_from_db() - assert job.status == TaskJob.Status.FINISHING - mock_finish.delay.assert_called_once_with(job.id) - - -def test_run_youtube_batch_skips_terminal_batch(mocker): - """A redelivered batch that already finished should not be re-run""" - mock_execute = mocker.patch( - "learning_resources.tasks._execute_youtube_batch", autospec=True - ) - mock_finish = mocker.patch( - "learning_resources.tasks.finish_youtube_etl", autospec=True - ) - - job = TaskJobFactory.create( - task_name=YOUTUBE_ETL_TASK_NAME, - status=TaskJob.Status.RUNNING, - params={"channel_ids": ["channel1"]}, - ) - batch = TaskBatchFactory.create(job=job, status=TaskBatch.Status.SUCCEEDED) - - run_youtube_batch(batch.id) - - mock_execute.assert_not_called() - # the redelivery still nudges the job to completion - job.refresh_from_db() - assert job.status == TaskJob.Status.FINISHING - mock_finish.delay.assert_called_once_with(job.id) - - -def test_run_youtube_batch_finishes_job_only_when_all_batches_done(mocker): - """The finish step should wait for every batch of the job""" - mocker.patch("learning_resources.tasks._execute_youtube_batch", autospec=True) - mock_finish = mocker.patch( - "learning_resources.tasks.finish_youtube_etl", autospec=True - ) - - job = TaskJobFactory.create( - task_name=YOUTUBE_ETL_TASK_NAME, - status=TaskJob.Status.RUNNING, - params={"channel_ids": ["channel1"]}, - ) - batches = TaskBatchFactory.create_batch(2, job=job) - - run_youtube_batch(batches[0].id) - mock_finish.delay.assert_not_called() - - run_youtube_batch(batches[1].id) - mock_finish.delay.assert_called_once_with(job.id) - - -@pytest.mark.parametrize("has_failures", [True, False]) -def test_finish_youtube_etl(mocker, has_failures): - """finish_youtube_etl should unpublish removed channels and close out the job""" - mock_unpublish = mocker.patch( - "learning_resources.tasks.loaders.unpublish_removed_youtube_channels", - autospec=True, - ) - mock_clear_cache = mocker.patch( - "learning_resources.tasks.clear_views_cache", autospec=True - ) - - job = TaskJobFactory.create( - task_name=YOUTUBE_ETL_TASK_NAME, - status=TaskJob.Status.FINISHING, - params={"channel_ids": ["channel1", "channel2"]}, - ) - TaskBatchFactory.create(job=job, status=TaskBatch.Status.SUCCEEDED) - if has_failures: - TaskBatchFactory.create( - job=job, - batch_key="channel:channel2", - status=TaskBatch.Status.FAILED, - error="boom", - ) - - finish_youtube_etl.delay(job.id) - - # cleanup runs either way; a failed batch leaves its resources as they were - mock_unpublish.assert_called_once_with(["channel1", "channel2"]) - mock_clear_cache.assert_called_once() - - job.refresh_from_db() - if has_failures: - assert job.status == TaskJob.Status.FAILED - assert "channel:channel2: boom" in job.error - else: - assert job.status == TaskJob.Status.SUCCEEDED - assert job.error == "" - - -def test_finish_youtube_etl_skips_unclaimed_job(mocker): - """finish_youtube_etl should do nothing for a job it hasn't claimed""" - mock_unpublish = mocker.patch( - "learning_resources.tasks.loaders.unpublish_removed_youtube_channels", - autospec=True, + mock_load_playlist = mocker.patch( + "learning_resources.tasks.loaders.load_playlist", autospec=True ) - job = TaskJobFactory.create( - task_name=YOUTUBE_ETL_TASK_NAME, - status=TaskJob.Status.SUCCEEDED, - params={"channel_ids": ["channel1"]}, + get_youtube_playlist_data.delay( + "channel1", _playlist_data("playlist1"), "ocw", create_videos=True ) - finish_youtube_etl.delay(job.id) - mock_unpublish.assert_not_called() + mock_load_playlist.assert_not_called() def test_get_youtube_transcripts(mocker): From 077fb70f1b32f7a7bbcaf33fa4e35e0008b222b9 Mon Sep 17 00:00:00 2001 From: Anastasia Beglova Date: Fri, 7 Aug 2026 11:22:10 -0400 Subject: [PATCH 3/5] commit --- learning_resources/etl/loaders.py | 106 ++++++++++++++++--------- learning_resources/etl/loaders_test.py | 51 ++++++++++-- learning_resources/tasks.py | 7 +- learning_resources/tasks_test.py | 23 +----- 4 files changed, 117 insertions(+), 70 deletions(-) diff --git a/learning_resources/etl/loaders.py b/learning_resources/etl/loaders.py index 046d59c81c..bbc1e0c2ce 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, @@ -1947,6 +1947,61 @@ def load_video_channel(video_channel_data: dict) -> VideoChannel: return video_channel +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 + ) + + 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 unpublish_playlists(playlist_resources: QuerySet) -> None: + """ + Unpublish playlist resources, and any video they leave orphaned + + Args: + playlist_resources (QuerySet): the playlist LearningResources to unpublish + """ + unpublished_ids = list(playlist_resources.values_list("id", flat=True)) + if not unpublished_ids: + return + + LearningResource.objects.filter(id__in=unpublished_ids).update(published=False) + bulk_resources_unpublished_actions( + unpublished_ids, LearningResourceType.video_playlist.name + ) + unpublish_orphaned_videos(unpublished_ids) + + def unpublish_removed_playlists( video_channel: VideoChannel, playlist_ids: list[str] ) -> None: @@ -1957,16 +2012,11 @@ def unpublish_removed_playlists( video_channel (VideoChannel): the video channel playlist_ids (list of str): youtube ids of the channel's current playlists """ - playlists_to_unpublish = LearningResource.objects.filter( - video_playlist__channel=video_channel - ).exclude(readable_id__in=playlist_ids) - unpublished_ids = list(playlists_to_unpublish.values_list("id", flat=True)) - - if unpublished_ids: - playlists_to_unpublish.update(published=False) - bulk_resources_unpublished_actions( - unpublished_ids, LearningResourceType.video_playlist.name + unpublish_playlists( + LearningResource.objects.filter(video_playlist__channel=video_channel).exclude( + readable_id__in=playlist_ids ) + ) def unpublish_removed_youtube_channels(channel_ids: list[str]) -> None: @@ -1984,35 +2034,17 @@ def unpublish_removed_youtube_channels(channel_ids: list[str]) -> None: ).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 + 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) ) - bulk_resources_unpublished_actions( - orphaned_playlist_ids, LearningResourceType.video_playlist.name - ) - - # 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 - ) + + # 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() def load_youtube_video_channels(video_channels_data: iter) -> list[VideoChannel]: diff --git a/learning_resources/etl/loaders_test.py b/learning_resources/etl/loaders_test.py index 8bdfc91211..2e94a22ac4 100644 --- a/learning_resources/etl/loaders_test.py +++ b/learning_resources/etl/loaders_test.py @@ -63,6 +63,7 @@ load_videos, load_videos_from_content_files, load_youtube_video_channels, + unpublish_orphaned_videos, unpublish_removed_playlists, ) from learning_resources.etl.mitxonline import transform_programs @@ -2868,16 +2869,27 @@ def test_load_playlists_unpublish(mocker): assert actual_unpublished_ids == expected_unpublished_ids -def test_unpublish_removed_playlists(mocker): - """Playlists no longer listed under the channel should be unpublished""" - mocker.patch("learning_resources_search.tasks.bulk_deindex_learning_resources.si") - mock_bulk_unpublish = mocker.patch( - "learning_resources.etl.loaders.bulk_resources_unpublished_actions", +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 ( @@ -2888,9 +2900,11 @@ def test_unpublish_removed_playlists(mocker): playlist.refresh_from_db() assert playlist.learning_resource.published is expected - mock_bulk_unpublish.assert_called_once_with( - [removed.learning_resource.id], LearningResourceType.video_playlist.name - ) + 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): @@ -2900,6 +2914,9 @@ def test_unpublish_removed_playlists_noop(mocker): ) channel = VideoChannelFactory.create() playlists = VideoPlaylistFactory.create_batch(2, channel=channel) + _add_playlist_videos( + playlists[0].learning_resource, [VideoFactory.create().learning_resource] + ) unpublish_removed_playlists( channel, [playlist.learning_resource.readable_id for playlist in playlists] @@ -2908,6 +2925,24 @@ def test_unpublish_removed_playlists_noop(mocker): mock_bulk_unpublish.assert_not_called() +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]) def test_load_ovs_playlist(mocker, playlist_exists, mock_get_similar_topics_qdrant): """Test load_ovs_playlist creates/updates a playlist and its videos""" diff --git a/learning_resources/tasks.py b/learning_resources/tasks.py index 651a870258..4510d278d4 100644 --- a/learning_resources/tasks.py +++ b/learning_resources/tasks.py @@ -586,9 +586,7 @@ def get_youtube_data(*, channel_ids=None): 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. Videos - # orphaned by playlists this run unpublishes are swept up by the next - # full 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] ) @@ -597,9 +595,6 @@ def get_youtube_data(*, channel_ids=None): for channel_config in channel_configs: get_youtube_channel_data.delay(channel_config) - # the fan-out writes after this point, but the views cache is short-lived - # and cleared again by the next ETL run, so there's nothing to wait for - clear_views_cache() return len(channel_configs) diff --git a/learning_resources/tasks_test.py b/learning_resources/tasks_test.py index 3e4f37d9eb..0a23c93899 100644 --- a/learning_resources/tasks_test.py +++ b/learning_resources/tasks_test.py @@ -64,23 +64,11 @@ def mock_blocklist(mocker): ) -def test_cache_is_cleared_after_task_run(mocker, mocked_celery, settings): +def test_cache_is_cleared_after_task_run(mocker, mocked_celery): """Test that the search cache is cleared out after every task run""" mocker.patch("learning_resources.tasks.ocw_courses_etl", autospec=True) mocker.patch("learning_resources.tasks.get_content_tasks", autospec=True) mocker.patch("learning_resources.tasks.pipelines") - settings.YOUTUBE_CONFIG_URL = "http://test.youtube/config.yaml" - settings.YOUTUBE_DEVELOPER_KEY = "key" - mocker.patch( - "learning_resources.tasks.youtube.get_youtube_channel_configs", - autospec=True, - return_value=[{"channel_id": "channel1"}], - ) - mocker.patch( - "learning_resources.tasks.loaders.unpublish_removed_youtube_channels", - autospec=True, - ) - mocker.patch("learning_resources.tasks.get_youtube_channel_data", autospec=True) mocked_clear_views_cache = mocker.patch( "learning_resources.tasks.clear_views_cache" ) @@ -98,9 +86,10 @@ def test_cache_is_cleared_after_task_run(mocker, mocked_celery, settings): 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): @@ -513,9 +502,6 @@ def test_get_youtube_data(mocker, youtube_settings, channel_ids): mock_channel_task = mocker.patch( "learning_resources.tasks.get_youtube_channel_data", autospec=True ) - mock_clear_cache = mocker.patch( - "learning_resources.tasks.clear_views_cache", autospec=True - ) assert get_youtube_data.delay(channel_ids=channel_ids).get() == 2 @@ -523,7 +509,6 @@ def test_get_youtube_data(mocker, youtube_settings, channel_ids): assert [ call.args[0] for call in mock_channel_task.delay.call_args_list ] == channel_configs - mock_clear_cache.assert_called_once() if channel_ids: # a run filtered to specific channels doesn't know the full channel set, From d0a844831a10550fd6795705d1c8ca0cc6e5ad9a Mon Sep 17 00:00:00 2001 From: Anastasia Beglova Date: Fri, 7 Aug 2026 11:41:18 -0400 Subject: [PATCH 4/5] remove no longer used code --- learning_resources/etl/loaders.py | 90 ----------------- learning_resources/etl/pipelines.py | 5 - learning_resources/etl/youtube.py | 133 ------------------------- learning_resources/etl/youtube_test.py | 95 ------------------ 4 files changed, 323 deletions(-) diff --git a/learning_resources/etl/loaders.py b/learning_resources/etl/loaders.py index bbc1e0c2ce..27d9c65e53 100644 --- a/learning_resources/etl/loaders.py +++ b/learning_resources/etl/loaders.py @@ -1870,44 +1870,6 @@ def load_videos_from_content_files( return videos -def load_playlists( - video_channel: VideoChannel, playlists_data: iter -) -> list[LearningResource]: - """ - 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 - - Returns: - list of LearningResource: - the created or updated LearningResources for the playlists - """ - 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] - - # 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, - ) - - return playlists - - def upsert_video_channel(video_channel_data: dict) -> VideoChannel: """ Create or update the VideoChannel row itself, without touching its playlists @@ -1930,23 +1892,6 @@ def upsert_video_channel(video_channel_data: dict) -> VideoChannel: return video_channel -def load_video_channel(video_channel_data: dict) -> VideoChannel: - """ - Load a single video channel, and its playlists, into the database - - Arg: - video_channel_data (dict): - the normalized video channel data - Returns: - VideoChannel: the updated or created video channel - """ - playlists_data = video_channel_data.pop("playlists", []) - video_channel = upsert_video_channel(video_channel_data) - load_playlists(video_channel, playlists_data) - - return video_channel - - def unpublish_orphaned_videos(playlist_ids: list[int] | None = None) -> None: """ Unpublish published videos that are in no published playlist. @@ -2045,38 +1990,3 @@ def unpublish_removed_youtube_channels(channel_ids: list[str]) -> None: # 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() - - -def load_youtube_video_channels(video_channels_data: iter) -> list[VideoChannel]: - """ - Load a list of video channels - - Args: - video_channels_data (iter of dict): iterable of the video channels data - - 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) - 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) - - unpublish_removed_youtube_channels(channel_ids) - - return video_channels 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 886f6910da..07e60362da 100644 --- a/learning_resources/etl/youtube.py +++ b/learning_resources/etl/youtube.py @@ -295,91 +295,6 @@ def extract_playlist_metadata( ) -def extract_playlists( - youtube_client: Resource, - playlist_configs: list[dict], - channel_id: str, - *, - create_videos_channel_setting: bool, -) -> Generator[tuple, None, None]: - """ - Extract a list of playlists for a channel, with 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 - """ - for playlist_data, create_videos in extract_playlist_metadata( - youtube_client, - playlist_configs, - channel_id, - create_videos_channel_setting=create_videos_channel_setting, - ): - yield ( - playlist_data, - extract_playlist_items(youtube_client, playlist_data["id"]), - create_videos, - ) - - -def extract_channels( - youtube_client: Resource, channels_config: list[dict] -) -> Generator[tuple, None, None]: - """ - Extract a list of channels - - Args: - youtube_client (Resource): Youtube api client - channels_config (list of dict): list of channel configurations - - Returns: - A generator that yields channel data - """ - 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, - ) - - 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}" - raise ExtractException(msg) from exc - - def extract_channel(youtube_client: Resource, channel_id: str) -> dict | None: """ Extract the raw data for a single channel @@ -513,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 @@ -625,33 +519,6 @@ def transform_channel(channel_data: dict) -> dict: } -def transform(extracted_channels: iter) -> Generator[dict, None, None]: - """ - Transform raw video data into normalized data structure - - Args: - extracted_channels (iterable of tuple): the youtube channels that were fetched - - 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 { - **transform_channel(channel_data), - # intentional generator expression - "playlists": ( - transform_playlist( - playlist, videos, offered_by, create_videos=create_videos - ) - for playlist, videos, create_videos in playlists - ), - } - - def get_youtube_videos_for_transcripts_job( *, created_after: str | None = None, diff --git a/learning_resources/etl/youtube_test.py b/learning_resources/etl/youtube_test.py index 046f075dd4..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"), @@ -473,23 +411,6 @@ def test_extract_playlists_create_videos( 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""" @@ -552,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"), [ From 2747e0db68f7b80365bae216418e8def00c63d1c Mon Sep 17 00:00:00 2001 From: Anastasia Beglova Date: Fri, 7 Aug 2026 12:02:34 -0400 Subject: [PATCH 5/5] fix test --- learning_resources/etl/loaders_test.py | 133 ++----------------------- 1 file changed, 7 insertions(+), 126 deletions(-) diff --git a/learning_resources/etl/loaders_test.py b/learning_resources/etl/loaders_test.py index 2e94a22ac4..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,9 +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 @@ -102,8 +100,6 @@ Program, TutorProblemFile, Video, - VideoChannel, - VideoPlaylist, ) from learning_resources.test_utils import set_up_topics from main.utils import now_in_utc @@ -2816,59 +2812,6 @@ 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") - 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, - ) - 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 - ) - - 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 - - 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 _add_playlist_videos(playlist_resource, videos): """Attach videos to a playlist resource""" playlist_resource.resources.set( @@ -3183,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 @@ -3269,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()