Isolate per-file extraction failures in canvas and edX content syncs

learning_resources/etl/canvas.py

@@ -60,25 +60,43 @@ def sync_canvas_archive(bucket, key: str, overwrite):
             overwrite=overwrite,
         )
         if run:
+            failed_content_keys = []
             canvas_content_files = list(
                 transform_canvas_content_files(
-                    course_archive_path, run, url_config=url_config, overwrite=overwrite
+                    course_archive_path,
+                    run,
+                    url_config=url_config,
+                    overwrite=overwrite,
+                    failed_keys=failed_content_keys,
                 )
             )
             content_files_ids = load_content_files(
                 run,
                 canvas_content_files,
+                failed_keys=failed_content_keys,
             )
 
+            failed_problem_paths = []
             canvas_problem_files = list(
                 transform_canvas_problem_files(
-                    course_archive_path, run, overwrite=overwrite
+                    course_archive_path,
+                    run,
+                    overwrite=overwrite,
+                    failed_source_paths=failed_problem_paths,
                 )
             )
-            problem_files_ids = load_problem_files(run, canvas_problem_files)
-            content_loaded = content_files_ids or not canvas_content_files
+            problem_files_ids = load_problem_files(
+                run,
+                canvas_problem_files,
+                failed_source_paths=failed_problem_paths,
+            )
+            content_loaded = content_files_ids or (
+                not canvas_content_files and not failed_content_keys
+            )
             # load_problem_file swallows per-file errors and returns None
-            problems_loaded = any(problem_files_ids) or not canvas_problem_files
+            problems_loaded = any(problem_files_ids) or (
+                not canvas_problem_files and not failed_problem_paths
+            )
             if content_loaded and problems_loaded:
                 # a failed or empty load must be retried on the next sync, so
                 # only mark processed once everything loaded (or was unpublished)
@@ -142,15 +160,25 @@ def run_for_canvas_archive(course_archive_path, course_folder, checksum, overwri
 
 
 def transform_canvas_content_files(
-    course_zipfile: Path, run: LearningResourceRun, url_config: dict, *, overwrite
+    course_zipfile: Path,
+    run: LearningResourceRun,
+    url_config: dict,
+    *,
+    overwrite,
+    failed_keys: list | None = None,
 ) -> Generator[dict, None, None]:
     """
     Transform published content files from a Canvas course zipfile
+
+    Files whose extraction fails are skipped and their existing records
+    are retained (not deleted/unpublished).
     """
     basedir = course_zipfile.name.split(".")[0]
     zipfile_path = course_zipfile.absolute()
     published_items = get_published_items(zipfile_path, url_config)
 
+    failed_source_paths = []
+
     def _generate_content():
         """Inner generator for yielding content data"""
         with (
@@ -166,7 +194,11 @@ def _generate_content():
                     log.debug("skipping unpublished file %s", member.filename)
 
             for content_data in process_olx_path(
-                olx_path, run, overwrite=overwrite, use_ocr=True
+                olx_path,
+                run,
+                overwrite=overwrite,
+                use_ocr=True,
+                failed_source_paths=failed_source_paths,
             ):
                 url_path = content_data["source_path"].lstrip(
                     content_data["source_path"].split("/")[0]
@@ -189,6 +221,13 @@ def _generate_content():
         full_path = Path(basedir) / Path(content_data["source_path"])
         published_keys.append(get_edx_module_id(str(full_path), run))
         yield content_data
+    # files whose extraction failed are retained, not treated as unpublished
+    for source_path in failed_source_paths:
+        full_path = Path(basedir) / Path(source_path)
+        failed_key = get_edx_module_id(str(full_path), run)
+        published_keys.append(failed_key)
+        if failed_keys is not None:
+            failed_keys.append(failed_key)
     unpublished_content = run.content_files.exclude(key__in=published_keys)
     # remove unpublished contentfiles
     bulk_resources_unpublished_actions(
@@ -198,10 +237,17 @@ def _generate_content():
 
 
 def transform_canvas_problem_files(
-    course_zipfile: Path, run: LearningResourceRun, *, overwrite
+    course_zipfile: Path,
+    run: LearningResourceRun,
+    *,
+    overwrite,
+    failed_source_paths: list | None = None,
 ) -> Generator[dict, None, None]:
     """
     Transform problem files from a Canvas course zipfile
+
+    Files whose extraction fails are skipped and their existing records
+    are retained (not deleted/unpublished).
     """
     basedir = course_zipfile.name.split(".")[0]
     with (
@@ -219,6 +265,7 @@ def transform_canvas_problem_files(
             valid_file_types=VALID_TUTOR_PROBLEM_FILE_TYPES,
             is_tutor_problem_file_import=True,
             use_ocr=True,
+            failed_source_paths=failed_source_paths,
         ):
             keys_to_keep = [
                 "run",

learning_resources/etl/canvas_test.py

@@ -37,7 +37,7 @@
     LearningResourceRunFactory,
     TutorProblemFileFactory,
 )
-from learning_resources.models import LearningResource
+from learning_resources.models import ContentFile, LearningResource
 from learning_resources_search.constants import CONTENT_FILE_TYPE
 from main.utils import now_in_utc
 
@@ -452,6 +452,91 @@ def test_transform_canvas_content_files_removes_unpublished_content(mocker, tmp_
     bulk_unpub.assert_called_once_with([unpublished_cf.id], CONTENT_FILE_TYPE)
 
 
+def test_transform_canvas_content_files_retains_failed_files(mocker, tmp_path):
+    """A file whose extraction raises must not be deleted; true orphans still are"""
+    resource = LearningResourceFactory.create(etl_source=ETLSource.canvas.name)
+    run = LearningResourceRunFactory.create(learning_resource=resource)
+
+    published_path = "/test/published/file1.html"
+    unpublished_path = "/test/unpublished/file2.html"
+    failing_cf = ContentFileFactory.create(
+        run=run, published=True, key=get_edx_module_id(published_path, run)
+    )
+    unpublished_cf = ContentFileFactory.create(
+        run=run, published=True, key=get_edx_module_id(unpublished_path, run)
+    )
+    module_xml = b"""<?xml version="1.0" encoding="UTF-8"?>
+    <modules xmlns="http://canvas.instructure.com/xsd/cccv1p0">
+      <module>
+        <title>Module 1</title>
+        <items>
+          <item>
+            <workflow_state>active</workflow_state>
+            <title>Item 1</title>
+            <identifierref>RES1</identifierref>
+            <content_type>resource</content_type>
+          </item>
+          <item>
+            <workflow_state>unpublished</workflow_state>
+            <title>Item 2</title>
+            <identifierref>RES2</identifierref>
+            <content_type>resource</content_type>
+          </item>
+        </items>
+      </module>
+    </modules>
+    """
+    manifest_xml = bytes(
+        f"""<?xml version="1.0" encoding="UTF-8"?>
+    <manifest xmlns="http://www.imsglobal.org/xsd/imsccv1p1/imscp_v1p1">
+      <resources>
+        <resource identifier="RES1" type="webcontent">
+          <file href="{published_path}"/>
+        </resource>
+        <resource identifier="RES2" type="webcontent">
+          <file href="{unpublished_path}"/>
+        </resource>
+      </resources>
+      <organizations>
+        <organization>
+          <item identifierref="RES1">
+            <title>Item 1</title>
+          </item>
+          <item identifierref="RES2">
+            <title>Item 2</title>
+          </item>
+        </organization>
+      </organizations>
+    </manifest>
+    """,
+        "utf-8",
+    )
+    zip_path = tmp_path / "canvas_course.zip"
+    with zipfile.ZipFile(zip_path, "w") as zf:
+        zf.writestr("course_settings/module_meta.xml", module_xml)
+        zf.writestr("imsmanifest.xml", manifest_xml)
+        zf.writestr(published_path, "content")
+        zf.writestr(unpublished_path, "content")
+
+    mocker.patch(
+        "learning_resources.etl.utils._extract_content",
+        side_effect=FileNotFoundError("ocr output missing"),
+    )
+    bulk_unpub = mocker.patch(
+        "learning_resources.etl.canvas.bulk_resources_unpublished_actions"
+    )
+
+    list(
+        transform_canvas_content_files(
+            Path(zip_path), run, url_config={}, overwrite=True
+        )
+    )
+
+    assert ContentFile.objects.filter(id=failing_cf.id).exists()
+    assert not ContentFile.objects.filter(id=unpublished_cf.id).exists()
+    bulk_unpub.assert_called_once_with([unpublished_cf.id], CONTENT_FILE_TYPE)
+
+
 @pytest.mark.parametrize("overwrite", [True, False])
 @pytest.mark.parametrize("existing_file", [True, False])
 def test_transform_canvas_problem_files_pdf_calls_pdf_to_markdown(  # noqa: PLR0913
@@ -2375,3 +2460,78 @@ def fake_download(key, dest):
         sync_mocks.bucket, "canvas/course_content/1/abc.imscc", overwrite=False
     )
     assert _canvas_run(readable_id).checksum
+
+
+TWO_FILE_MANIFEST_XML = b"""<?xml version="1.0" encoding="UTF-8"?>
+<manifest xmlns="http://www.imsglobal.org/xsd/imsccv1p1/imscp_v1p1">
+  <resources>
+    <resource identifier="RES3" type="webcontent">
+      <file href="web_resources/file3.html"/>
+    </resource>
+    <resource identifier="RES4" type="webcontent">
+      <file href="web_resources/file4.pdf"/>
+    </resource>
+  </resources>
+</manifest>
+"""
+
+
+def test_sync_canvas_archive_partial_failure_retains_and_stamps(
+    mocker, tmp_path, sync_mocks
+):
+    """One raising file: others yield, failed record survives the transform's
+    delete pass, failed_keys reaches load_content_files, checksum is stamped
+    """
+    two_file_zip = make_timed_lock_zip(
+        tmp_path,
+        "2001-01-01T00:00:00",
+        name="two_files.zip",
+        manifest_xml=TWO_FILE_MANIFEST_XML,
+    )
+    with zipfile.ZipFile(two_file_zip, "a") as zf:
+        zf.writestr("web_resources/file4.pdf", b"%PDF-fake")
+
+    def fake_download(_key, dest):
+        Path(dest).write_bytes(two_file_zip.read_bytes())
+
+    sync_mocks.bucket.download_file.side_effect = fake_download
+
+    # first sync materializes the resource/run (loaders are mocked, so no rows)
+    key = "canvas/course_content/1/abc.imscc"
+    readable_id = sync_canvas_archive(sync_mocks.bucket, key, overwrite=False)
+    run = _canvas_run(readable_id)
+    failing_key = get_edx_module_id(str(Path("abc") / "web_resources/file4.pdf"), run)
+    existing = ContentFileFactory.create(run=run, key=failing_key, published=True)
+
+    def fake_extract(document, metadata, olx_path, key, **kwargs):
+        if "file4.pdf" 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
+    )
+
+    sync_canvas_archive(sync_mocks.bucket, key, overwrite=True)
+
+    assert ContentFile.objects.filter(id=existing.id).exists()
+    assert sync_mocks.load_content.call_args.kwargs["failed_keys"] == [failing_key]
+    assert _canvas_run(readable_id).checksum
+
+
+def test_sync_canvas_archive_total_failure_does_not_stamp_checksum(mocker, sync_mocks):
+    """Every file raising: checksum NOT stamped, so next sync retries"""
+    # a bare MagicMock return is truthy and would satisfy content_loaded via
+    # content_files_ids, bypassing the gate under test
+    sync_mocks.load_content.return_value = []
+    mocker.patch(
+        "learning_resources.etl.utils._extract_content",
+        side_effect=FileNotFoundError("converter output missing"),
+    )
+
+    readable_id = sync_canvas_archive(
+        sync_mocks.bucket, "canvas/course_content/1/abc.imscc", overwrite=False
+    )
+
+    assert _canvas_run(readable_id).checksum is None

learning_resources/etl/edx_shared.py

@@ -148,17 +148,23 @@ def process_course_archive(
             log.info("Checksums match for %s, skipping load", key)
             return True
         try:
+            failed_keys = []
             content_files_data = iter(
-                transform_content_files(course_tarpath, run, overwrite=overwrite)
+                transform_content_files(
+                    course_tarpath, run, overwrite=overwrite, failed_keys=failed_keys
+                )
             )
             first = next(content_files_data, None)
             if first is None:
+                if failed_keys:
+                    # every file failed: retry next sync, don't mark as empty
+                    return True
                 # empty archive: stop re-downloading it
                 run.archive_key = key
                 run.save(update_fields=["archive_key"])
                 return True
             content_files_ids = load_content_files(
-                run, chain([first], content_files_data)
+                run, chain([first], content_files_data), failed_keys=failed_keys
             )
             if content_files_ids:
                 run.checksum = checksum