fix(transform): fall back to pdfa when primary PDF is empty

cds_migrator_kit/rdm/records/load/entities/record.py

@@ -8,6 +8,7 @@
 """Creates, versions, and publishes a single RDM record from a ``RecordEntry``."""
 
 import os
+import re
 from typing import Dict
 
 import arrow
@@ -33,6 +34,12 @@
 
 from .parent import ParentLoad
 
+# /Type /Pages … /Count 0 inside one object (before endobj).
+_PDF_ZERO_PAGES_RE = re.compile(
+    rb"/Type\s*/Pages\b(?:(?!endobj).)*?/Count\s+0\b",
+    re.DOTALL,
+)
+
 
 def import_legacy_files(filepath):
     """Download file from legacy."""
@@ -45,6 +52,23 @@ def import_legacy_files(filepath):
     return filestream
 
 
+def file_is_empty(filepath):
+    """Return True if the file on disk has zero bytes."""
+    return os.path.getsize(filepath) == 0
+
+
+def pdf_has_zero_pages(filepath):
+    """Return True if the file is a PDF whose /Pages dictionary has /Count 0.
+
+    Detects PDFs from the ``%PDF`` magic header on disk, not legacy mimetype.
+    """
+    with open(filepath, "rb") as stream:
+        content = stream.read()
+    if not content.startswith(b"%PDF"):
+        return False
+    return _PDF_ZERO_PAGES_RE.search(content) is not None
+
+
 class RecordLoad:
     """Creates, versions, and publishes a single RDM record from a ``RecordEntry``.
 
@@ -118,6 +142,27 @@ def load_files(
                     ],
                     uow=uow,
                 )
+                path = file_data["eos_tmp_path"]
+                if file_is_empty(path):
+                    raise ManualImportRequired(
+                        message="Empty file detected; curator review required.",
+                        field="files",
+                        stage="load",
+                        recid=recid,
+                        priority="critical",
+                        value=file_data["key"],
+                    )
+                if pdf_has_zero_pages(path):
+                    raise ManualImportRequired(
+                        message=(
+                            "Empty/corrupt PDF detected; curator review required."
+                        ),
+                        field="files",
+                        stage="load",
+                        recid=recid,
+                        priority="critical",
+                        value=file_data["key"],
+                    )
                 current_rdm_records_service.draft_files.set_file_content(
                     identity,
                     draft.id,

cds_migrator_kit/rdm/records/transform/config.py

@@ -159,6 +159,9 @@
 
 FILE_SUBFORMATS_TO_DROP = ["pdfa", "unstamped"]
 
+# Known empty/corrupt legacy PDF stub (Ghostscript zero-page placeholder).
+EMPTY_PDF_CHECKSUM = "6f58aa0bfac8a909ac578c1595d55a60"
+
 # Legacy experiment names remapped to vocabulary ids before lookup
 EXPERIMENT_ALIASES = {
     "t2k": "re13",

cds_migrator_kit/rdm/records/transform/transform_versions.py

@@ -11,7 +11,11 @@
 
 import arrow
 
-from cds_migrator_kit.rdm.records.transform.config import FILE_SUBFORMATS_TO_DROP
+from cds_migrator_kit.errors import ManualImportRequired
+from cds_migrator_kit.rdm.records.transform.config import (
+    EMPTY_PDF_CHECKSUM,
+    FILE_SUBFORMATS_TO_DROP,
+)
 from cds_migrator_kit.rdm.records.transform.entities.version import RecordVersion
 
 
@@ -62,6 +66,15 @@ def build(self):
         for file_dump in self.raw_dump_entry["files"]:
             if self._should_skip_file(file_dump):
                 continue
+            if self._is_empty_or_corrupt_file(file_dump):
+                raise ManualImportRequired(
+                    message="Empty/corrupt PDF detected; curator review required.",
+                    field="files",
+                    value=file_dump["full_name"],
+                    stage="transform",
+                    recid=str(file_dump["recid"]),
+                    priority="critical",
+                )
             version_number = file_dump["version"]
             raw_file_dumps_by_version.setdefault(version_number, []).append(file_dump)
             representative_file.setdefault(version_number, file_dump)
@@ -138,6 +151,11 @@ def _is_external_doi(self):
         doi = self.record.body.get("pids", {}).get("doi", {})
         return doi.get("provider") == "external"
 
+    @staticmethod
+    def _is_empty_or_corrupt_file(file_dump):
+        """Return True for the known empty/corrupt Ghostscript PDF stub."""
+        return file_dump["checksum"] == EMPTY_PDF_CHECKSUM
+
     def _should_skip_file(self, file_dump):
         if file_dump["subformat"] in FILE_SUBFORMATS_TO_DROP:
             self.migration_logger.add_information(

tests/cds-rdm/test_transform_versions.py

@@ -13,6 +13,11 @@
 import pytest
 
 from cds_migrator_kit.errors import ManualImportRequired
+from cds_migrator_kit.rdm.records.load.entities.record import (
+    file_is_empty,
+    pdf_has_zero_pages,
+)
+from cds_migrator_kit.rdm.records.transform.config import EMPTY_PDF_CHECKSUM
 from cds_migrator_kit.rdm.records.transform.transform import CDSToRDMRecordTransform
 
 
@@ -396,3 +401,45 @@ def test_versions_restriction_lifted_on_only_file_becomes_public(transform):
         "access_obj": {"record": "public", "files": "public"}
     }
     assert versions[2]["files"]["A.pdf"]["version"] == 2
+
+
+def test_known_corrupt_checksum_raises_manual_import(transform):
+    """Known empty/corrupt stub checksum fails for curator review."""
+    raw_dump_entry = {
+        "recid": 123,
+        "files": [
+            _file_dump(
+                full_name="cer-002523122.pdf",
+                checksum=EMPTY_PDF_CHECKSUM,
+            ),
+            # pdfa sibling is still dropped; we fail instead of falling back to it
+            _file_dump(
+                full_name="cer-002523122.pdf",
+                file_version=2,
+                subformat="pdfa",
+                checksum="valid-pdfa",
+            ),
+        ],
+    }
+
+    with pytest.raises(ManualImportRequired) as exc_info:
+        transform._versions(raw_dump_entry, _record())
+
+    assert exc_info.value.recid == "123"
+    assert exc_info.value.value == "cer-002523122.pdf"
+    assert "curator review required" in exc_info.value.message
+
+
+def test_load_detects_empty_file_and_pdf_from_disk(tmp_path):
+    """Load checks use the file on disk, not legacy mimetype metadata."""
+    empty = tmp_path / "empty.txt"
+    empty.write_bytes(b"")
+    pdf = tmp_path / "stub.bin"
+    pdf.write_bytes(b"%PDF-1.4\n1 0 obj\n<< /Type /Pages /Count 0 >>\nendobj\n")
+    not_pdf = tmp_path / "notes.pdf"
+    not_pdf.write_bytes(b"plain text")
+
+    assert file_is_empty(empty) is True
+    assert file_is_empty(pdf) is False
+    assert pdf_has_zero_pages(pdf) is True
+    assert pdf_has_zero_pages(not_pdf) is False