good
") + (static_dir / "bad.html").write_text("bad
") + tarball = tmp_path / "course.tar.gz" + with tarfile.open(tarball, "w:gz") as tar: + tar.add(course_dir, arcname="course") + return tarball + + +def test_process_course_archive_retains_failed_file(mocker, tmp_path): + """One raising file: others load, its existing ContentFile stays published, + checksum and archive_key are stamped + """ + # load_content_files no-ops for non-course resources; pin the random factory + run = LearningResourceRunFactory.create( + archive_key=None, + checksum=None, + learning_resource=LearningResourceFactory.create(is_course=True), + ) + tarball = _make_olx_tarball(tmp_path) + bucket = mocker.MagicMock() + bucket.download_file.side_effect = lambda _key, dest: shutil.copy(tarball, dest) + + failing_key = get_edx_module_id("course/static/bad.html", run) + existing = ContentFileFactory.create(run=run, key=failing_key, published=True) + + def fake_extract(document, metadata, olx_path, key, **kwargs): + if "bad.html" in metadata["source_path"]: + msg = "converter output missing" + raise FileNotFoundError(msg) + return {"content": "text", "content_title": ""} + + mocker.patch( + "learning_resources.etl.utils._extract_content", side_effect=fake_extract + ) + # search-index plugin hook, not under test; depends on randomized resource + # fields and hits a nonexistent opensearch in CI + mocker.patch("learning_resources.etl.loaders.content_files_loaded_actions") + + key = "20240101/courses/course.tar.gz" + process_course_archive(bucket, key, run) + + good_key = get_edx_module_id("course/static/good.html", run) + assert ContentFile.objects.filter(run=run, key=good_key).exists() + existing.refresh_from_db() + assert existing.published is True + run.refresh_from_db() + assert run.checksum + assert run.archive_key == key + + +def test_process_course_archive_all_failures_not_marked_empty(mocker, tmp_path): + """Every file raising: archive_key is NOT stamped, so the archive retries""" + run = LearningResourceRunFactory.create( + archive_key=None, + checksum=None, + learning_resource=LearningResourceFactory.create(is_course=True), + ) + tarball = _make_olx_tarball(tmp_path) + bucket = mocker.MagicMock() + bucket.download_file.side_effect = lambda _key, dest: shutil.copy(tarball, dest) + + mocker.patch( + "learning_resources.etl.utils._extract_content", + side_effect=FileNotFoundError("converter output missing"), + ) + + process_course_archive(bucket, "20240101/courses/course.tar.gz", run) + + run.refresh_from_db() + assert run.archive_key is None + assert run.checksum is None diff --git a/learning_resources/etl/loaders.py b/learning_resources/etl/loaders.py index 0460777b29..1c1f045fba 100644 --- a/learning_resources/etl/loaders.py +++ b/learning_resources/etl/loaders.py @@ -1014,6 +1014,7 @@ def load_content_files( content_files_data: list[dict], *, calc_completeness: bool = False, + failed_keys: list | None = None, ) -> list[int]: """ Sync all content files for a course run to database and S3 if not present in DB @@ -1022,6 +1023,9 @@ def load_content_files( course_run (LearningResourceRun): a course run content_files_data (list or generator): Details about the content files calc_completeness: bool: Whether to calculate the completeness score + failed_keys: list: caller-owned list of ContentFile keys whose + extraction failed, mutated in place while content_files_data is + consumed; those records are exempted from the stale/unpublish pass Returns: list of int: Ids of the ContentFile objects that were created/updated @@ -1047,6 +1051,8 @@ def load_content_files( stale_published_files = ContentFile.objects.filter( run=course_run, published=True ).exclude(id__in=content_files_ids) + if failed_keys: + stale_published_files = stale_published_files.exclude(key__in=failed_keys) stale_direct_resource_ids = list( stale_published_files.filter(direct_learning_resource__isnull=False) .values_list("direct_learning_resource_id", flat=True) @@ -1198,6 +1204,8 @@ def load_problem_file( def load_problem_files( course_run: LearningResourceRun, problem_files_data: list[dict], + *, + failed_source_paths: list | None = None, ) -> list[int]: """ Sync all problem files for canvas course @@ -1205,6 +1213,9 @@ def load_problem_files( Args: course_run (LearningResourceRun): a course run problem_files_data (list or generator): Details about the problem files + failed_source_paths (list): caller-owned list of source paths whose + extraction failed, mutated in place while problem_files_data is + consumed; those records are retained rather than deleted as orphans Returns: list of int: Ids of the TutorProblemFile objects that were created/updated @@ -1214,11 +1225,12 @@ def load_problem_files( load_problem_file(course_run, problem_file) for problem_file in problem_files_data ] - for file in ( - TutorProblemFile.objects.filter(run=course_run) - .exclude(id__in=problem_files_ids) - .all() - ): + deletable = TutorProblemFile.objects.filter(run=course_run).exclude( + id__in=problem_files_ids + ) + if failed_source_paths: + deletable = deletable.exclude(source_path__in=failed_source_paths) + for file in deletable.all(): file.delete() return problem_files_ids diff --git a/learning_resources/etl/loaders_test.py b/learning_resources/etl/loaders_test.py index f6e254c280..caaa4bee71 100644 --- a/learning_resources/etl/loaders_test.py +++ b/learning_resources/etl/loaders_test.py @@ -1993,6 +1993,46 @@ def test_load_content_files_does_not_update_already_unpublished_stale_files(mock assert stale_unpublished_file.updated_on == old_timestamp +def test_load_content_files_failed_keys_stay_published(mocker): + """Contentfiles whose extraction failed are not unpublished as stale""" + course = LearningResourceFactory.create(is_course=True, create_runs=False) + course_run = LearningResourceRunFactory.create( + published=True, learning_resource=course + ) + loaded_cf = ContentFileFactory.create( + run=course_run, published=True, key="loaded-key" + ) + failed_cf = ContentFileFactory.create( + run=course_run, published=True, key="failed-key" + ) + stale_cf = ContentFileFactory.create( + run=course_run, published=True, key="stale-key" + ) + + mocker.patch( + "learning_resources.etl.loaders.load_content_file", + return_value=loaded_cf.id, + autospec=True, + ) + mocker.patch( + "learning_resources.etl.loaders.content_files_loaded_actions", + autospec=True, + ) + + result = load_content_files( + course_run, + [{"key": "loaded-key", "content": "text"}], + failed_keys=["failed-key"], + ) + + assert result == [loaded_cf.id] + + failed_cf.refresh_from_db() + stale_cf.refresh_from_db() + assert failed_cf.published is True + assert stale_cf.published is False + + @pytest.mark.parametrize("test_mode", [True, False]) def test_load_test_mode_resource_content_files( mocker, mock_course_archive_bucket, test_mode @@ -2140,6 +2180,22 @@ def test_load_problem_files(mocker): ).exists() +def test_load_problem_files_retains_failed_source_paths(): + """Problem files whose extraction failed are not deleted as orphans""" + run = LearningResourceRunFactory.create() + failed = TutorProblemFileFactory.create( + run=run, source_path="web_resources/ai/tutor/p1/failed.pdf" + ) + orphan = TutorProblemFileFactory.create( + run=run, source_path="web_resources/ai/tutor/p2/orphan.pdf" + ) + + load_problem_files(run, [], failed_source_paths=[failed.source_path]) + + assert TutorProblemFile.objects.filter(id=failed.id).exists() + assert not TutorProblemFile.objects.filter(id=orphan.id).exists() + + def test_load_image(): """Test that image resources are uniquely created or retrieved based on parameters""" resource_url = "https://mit.edu" diff --git a/learning_resources/etl/utils.py b/learning_resources/etl/utils.py index d06c1628ad..0292b5018e 100644 --- a/learning_resources/etl/utils.py +++ b/learning_resources/etl/utils.py @@ -66,6 +66,10 @@ log = logging.getLogger(__name__) +class InvalidPDFError(Exception): + """Raised when a PDF fails pypdf validation before extraction.""" + + def load_offeror_topic_map(offeror_code: str): """ Load the topic mappings from the database. @@ -648,12 +652,16 @@ def _extract_content( # noqa: PLR0913 file_extension = metadata.get("file_extension") file_path = Path(olx_path) / Path(source_path) if file_extension == ".pdf" and file_path.is_file() and not pdf_is_valid(file_path): - log.warning("Skipping invalid pdf %s", file_path) - return None + msg = f"Invalid PDF {file_path}" + raise InvalidPDFError(msg) if _should_use_ocr( file_extension=file_extension, file_path=file_path, use_ocr=use_ocr ): - content_dict = _extract_content_with_ocr(file_path, is_tutor_problem) + try: + content_dict = _extract_content_with_ocr(file_path, is_tutor_problem) + except Exception: # noqa: BLE001 + log.warning("OCR extraction failed for %s, falling back to tika", file_path) + content_dict = None if content_dict: return content_dict @@ -798,8 +806,15 @@ def process_olx_path( # noqa: PLR0913 valid_file_types=VALID_TEXT_FILE_TYPES, is_tutor_problem_file_import=False, use_ocr=False, + failed_source_paths: list | None = None, ) -> Generator[dict, None, None]: - """Process OLX path and yield content dictionaries.""" + """ + Process OLX path and yield content dictionaries. + + failed_source_paths is a caller-owned list appended to in place with the + source_path of each file whose processing raises; it is only complete once + the generator is fully exhausted. + """ video_srt_metadata = get_video_metadata(olx_path, run) for document, metadata in documents_from_olx( @@ -808,25 +823,35 @@ def process_olx_path( # noqa: PLR0913 source_path = metadata.get("source_path") key = get_edx_module_id(source_path, run) - existing_record = _get_existing_record( - source_path, key, run, is_tutor_problem_file_import - ) - - if _should_reprocess(existing_record, metadata, overwrite): - content_dict = _extract_content( - document, - metadata, - olx_path, - key, - use_ocr=use_ocr, - is_tutor_problem=is_tutor_problem_file_import, + try: + existing_record = _get_existing_record( + source_path, key, run, is_tutor_problem_file_import ) - if content_dict is None: - continue - else: - content_dict = _get_cached_content( - existing_record, is_tutor_problem_file_import + + if _should_reprocess(existing_record, metadata, overwrite): + content_dict = _extract_content( + document, + metadata, + olx_path, + key, + use_ocr=use_ocr, + is_tutor_problem=is_tutor_problem_file_import, + ) + if content_dict is None: + continue + else: + content_dict = _get_cached_content( + existing_record, is_tutor_problem_file_import + ) + except Exception: + log.exception( + "Extraction failed for %s in run %s, skipping file", + source_path, + run.id, ) + if failed_source_paths is not None: + failed_source_paths.append(source_path) + continue yield _build_result( olx_path, metadata, key, run, video_srt_metadata, content_dict @@ -834,7 +859,11 @@ def process_olx_path( # noqa: PLR0913 def transform_content_files( - course_tarpath: Path, run: LearningResourceRun, *, overwrite: bool + course_tarpath: Path, + run: LearningResourceRun, + *, + overwrite: bool, + failed_keys: list | None = None, ) -> Generator[dict, None, None]: """ Pass content to tika, then return JSON document with transformed content inside it @@ -842,15 +871,25 @@ def transform_content_files( Args: course_tarpath (str): The path to the tarball which contains the OLX run (LearningResourceRun): The run associated witb the content files + failed_keys (list): caller-owned list extended in place with the + ContentFile keys of files whose extraction failed; valid only once + the generator is fully exhausted Yields: dict: content from file """ basedir = course_tarpath.name.split(".")[0] + failed_source_paths = [] with TemporaryDirectory(prefix=basedir) as inner_tempdir: check_call(["tar", "xf", course_tarpath], cwd=inner_tempdir) # noqa: S603,S607 olx_path = glob.glob(inner_tempdir + "/*")[0] # noqa: PTH207 - yield from process_olx_path(olx_path, run, overwrite=overwrite) + yield from process_olx_path( + olx_path, run, overwrite=overwrite, failed_source_paths=failed_source_paths + ) + if failed_keys is not None: + failed_keys.extend( + get_edx_module_id(source_path, run) for source_path in failed_source_paths + ) def get_s3_prefix_for_source(etl_source: str) -> str: diff --git a/learning_resources/etl/utils_test.py b/learning_resources/etl/utils_test.py index 58704fd55d..9dba4f2f3b 100644 --- a/learning_resources/etl/utils_test.py +++ b/learning_resources/etl/utils_test.py @@ -949,7 +949,7 @@ def test_process_olx_path_malformed_sjson(mocker, settings): def test_process_olx_path_encrypted_pdf(mocker, settings, tmp_path): """ - Test that process_olx_path logs an error and skips encrypted PDFs + Test that process_olx_path records encrypted PDFs as failures and skips them """ settings.OCR_MODEL = "test_model" settings.SKIP_TIKA = False @@ -998,17 +998,22 @@ def test_process_olx_path_encrypted_pdf(mocker, settings, tmp_path): side_effect=pypdf.errors.FileNotDecryptedError, ) + failed = [] results = list( utils.process_olx_path( str(olx_path), run, overwrite=True, use_ocr=True, + failed_source_paths=failed, ) ) assert len(results) == 0 - mock_log.warning.assert_called_with("Skipping invalid pdf %s", full_path) + assert failed == [source_rel_path] + mock_log.exception.assert_called_with( + "Extraction failed for %s in run %s, skipping file", source_rel_path, run.id + ) tika_from_buffer_mock.assert_not_called() @@ -1150,3 +1155,85 @@ def test_extract_content_ocr_fallback_to_tika(mocker, settings, tmp_path): ) assert result == {"content": "tika content", "content_title": "Tika Title"} + + +def test_extract_content_ocr_failure_falls_back_to_tika( + mocker, settings, tmp_path, caplog +): + """An OCR crash (e.g. missing converter output JSON) should fall through to tika""" + settings.SKIP_TIKA = False + mocker.patch("learning_resources.etl.utils._should_use_ocr", return_value=True) + mocker.patch( + "learning_resources.etl.utils._extract_content_with_ocr", + side_effect=FileNotFoundError("no converter output json"), + ) + mocker.patch( + "learning_resources.etl.utils.extract_text_metadata", + return_value={"content": "tika text", "metadata": {"title": "doc title"}}, + ) + result = utils._extract_content( # noqa: SLF001 + b"pdf bytes", + { + "source_path": "web_resources/7.06_Fall2025_Exam1_answers.pdf", + "file_extension": ".pdf", + "mime_type": "application/pdf", + }, + str(tmp_path), + "test-key", + use_ocr=True, + ) + assert result["content"] == "tika text" + ocr_records = [r for r in caplog.records if "OCR extraction failed" in r.message] + assert ocr_records + assert all(r.levelname == "WARNING" for r in ocr_records) + + +@pytest.mark.django_db +def test_process_olx_path_skips_failed_files(mocker, tmp_path): + """A file whose extraction raises is skipped and recorded; other files still yield""" + run = LearningResourceRunFactory.create() + static_dir = tmp_path / "static" + static_dir.mkdir() + (static_dir / "good.html").write_text("good
") + (static_dir / "bad.html").write_text("bad
") + + def fake_extract(document, metadata, olx_path, key, **kwargs): + if "bad.html" in metadata["source_path"]: + msg = "converter output missing" + raise FileNotFoundError(msg) + return {"content": "text", "content_title": ""} + + mocker.patch( + "learning_resources.etl.utils._extract_content", side_effect=fake_extract + ) + failed = [] + results = list( + utils.process_olx_path( + str(tmp_path), run, overwrite=True, failed_source_paths=failed + ) + ) + assert len(results) == 1 + assert "good.html" in results[0]["source_path"] + assert len(failed) == 1 + assert "bad.html" in failed[0] + + +def test_extract_content_invalid_pdf_raises(mocker, settings, tmp_path): + """A PDF failing pdf_is_valid raises, so process_olx_path records a failure + instead of the old silent drop + """ + settings.SKIP_TIKA = False # _extract_content short-circuits before pdf_is_valid + (tmp_path / "bad.pdf").write_bytes(b"%PDF- not really") + mocker.patch("learning_resources.etl.utils.pdf_is_valid", return_value=False) + + with pytest.raises(utils.InvalidPDFError): + utils._extract_content( # noqa: SLF001 + b"%PDF- not really", + { + "source_path": "bad.pdf", + "file_extension": ".pdf", + "mime_type": "application/pdf", + }, + str(tmp_path), + "test-key", + )