diff --git a/integrations/opendataloader_pdf/src/haystack_integrations/components/converters/opendataloader_pdf/converter.py b/integrations/opendataloader_pdf/src/haystack_integrations/components/converters/opendataloader_pdf/converter.py index a128304581..f10e4cf2ca 100644 --- a/integrations/opendataloader_pdf/src/haystack_integrations/components/converters/opendataloader_pdf/converter.py +++ b/integrations/opendataloader_pdf/src/haystack_integrations/components/converters/opendataloader_pdf/converter.py @@ -54,6 +54,7 @@ def __init__( convert_kwargs: dict[str, Any] | None = None, extract_images: bool = False, image_output_dir: str | Path | None = None, + skip_failed_pdfs: bool = False, ) -> None: """ Initialize the OpenDataLoader converter. @@ -65,6 +66,8 @@ def __init__( :param extract_images: Whether to extract images and return them through the `image_documents` output. :param image_output_dir: Persistent root directory for extracted image files. Each `run()` stores its images in a unique subdirectory of this directory. Required when `extract_images` is `True`. + :param skip_failed_pdfs: Bisect failed batches and skip failing PDFs with a warning. Java and source + preparation errors still propagate. :raises ValueError: If image extraction is enabled without an output directory. """ conversion_options = convert_kwargs.copy() if convert_kwargs else {} @@ -83,6 +86,7 @@ def __init__( self.convert_kwargs = conversion_options self.extract_images = extract_images self.image_output_dir = Path(image_output_dir) if image_output_dir is not None else None + self.skip_failed_pdfs = skip_failed_pdfs def to_dict(self) -> dict[str, Any]: """ @@ -97,6 +101,7 @@ def to_dict(self) -> dict[str, Any]: convert_kwargs=self.convert_kwargs, extract_images=self.extract_images, image_output_dir=str(self.image_output_dir) if self.image_output_dir is not None else None, + skip_failed_pdfs=self.skip_failed_pdfs, ) @classmethod @@ -188,6 +193,63 @@ def _read_output(self, output_dir: Path, pdf_path: Path) -> str: raise RuntimeError(message) return output_file.read_text(encoding="utf-8") + def _convert_batch( + self, + sources: list[tuple[Path, dict[str, Any]]], + output_dir: Path, + image_dir: Path | None, + ) -> list[Document]: + """Convert a non-empty batch, bisecting failures into isolated output directories when enabled.""" + output_dir.mkdir(parents=True, exist_ok=True) + conversion_kwargs = {"image_output": "off", **self.convert_kwargs} + if image_dir is not None: + image_dir.mkdir(parents=True, exist_ok=True) + conversion_kwargs.update(image_output="external", image_dir=str(image_dir)) + + try: + opendataloader_pdf.convert( + input_path=[str(pdf_path) for pdf_path, _ in sources], + output_dir=str(output_dir), + format=self.output_format, + **conversion_kwargs, + ) + contents = [self._read_output(output_dir=output_dir, pdf_path=pdf_path) for pdf_path, _ in sources] + except (subprocess.CalledProcessError, RuntimeError) as exc: + if not self.skip_failed_pdfs: + raise + # Failed attempts may leave seemingly valid text and images. Never reuse their outputs. + shutil.rmtree(output_dir) + if image_dir is not None: + shutil.rmtree(image_dir) + if len(sources) == 1: + pdf_path, document_meta = sources[0] + logger.warning( + "Skipping PDF {file_path} after conversion failure: {error}", + file_path=document_meta.get("file_path", str(pdf_path)), + error=exc, + ) + return [] + + logger.warning( + "Batch of {count} PDFs failed; splitting into smaller batches: {error}", count=len(sources), error=exc + ) + midpoint = len(sources) // 2 + documents = [] + for name, batch in (("left", sources[:midpoint]), ("right", sources[midpoint:])): + documents.extend( + self._convert_batch( + sources=batch, + output_dir=output_dir / name, + image_dir=image_dir / name if image_dir is not None else None, + ) + ) + return documents + + return [ + Document(content=content, meta={**document_meta, "output_format": self.output_format}) + for (_, document_meta), content in zip(sources, contents, strict=True) + ] + @component.output_types(documents=list[Document], image_documents=list[Document]) def run( self, @@ -202,7 +264,8 @@ def run( source. A list must contain one dictionary per source. ByteStream metadata is also preserved. :returns: Dictionary containing the converted text Documents and image Documents. Each image Document has the - persistent extracted image path in its `file_path` metadata field. + persistent extracted image path in its `file_path` metadata field. With `skip_failed_pdfs=True`, PDFs + that fail individual conversion are omitted from both outputs. """ if not sources: return {"documents": [], "image_documents": []} @@ -226,19 +289,7 @@ def run( output_dir.mkdir() prepared_sources = self._prepare_sources(sources=sources, metadata=metadata, input_dir=input_dir) - conversion_kwargs = {"image_output": "off", **self.convert_kwargs} - if image_dir is not None: - conversion_kwargs.update(image_output="external", image_dir=str(image_dir)) - - opendataloader_pdf.convert( - input_path=[str(pdf_path) for pdf_path, _ in prepared_sources], - output_dir=str(output_dir), - format=self.output_format, - **conversion_kwargs, - ) - for pdf_path, document_meta in prepared_sources: - content = self._read_output(output_dir=output_dir, pdf_path=pdf_path) - documents.append(Document(content=content, meta={**document_meta, "output_format": self.output_format})) + documents = self._convert_batch(sources=prepared_sources, output_dir=output_dir, image_dir=image_dir) if image_dir is not None: image_documents = [ diff --git a/integrations/opendataloader_pdf/tests/test_converter.py b/integrations/opendataloader_pdf/tests/test_converter.py index 190a282247..e48b625e67 100644 --- a/integrations/opendataloader_pdf/tests/test_converter.py +++ b/integrations/opendataloader_pdf/tests/test_converter.py @@ -1,3 +1,7 @@ +# SPDX-FileCopyrightText: 2026-present deepset GmbH +# +# SPDX-License-Identifier: Apache-2.0 + import subprocess from concurrent.futures import ThreadPoolExecutor from pathlib import Path @@ -429,6 +433,152 @@ def fail_java_check(): assert not (tmp_path / "images").exists() +def test_converter_bisects_failed_batches_and_cleans_partial_outputs( + tmp_path, _mock_opendataloader, monkeypatch, caplog +): + files = [tmp_path / name for name in ("good.pdf", "bad.pdf", "another.pdf", "fourth.pdf", "fifth.pdf")] + for pdf in files: + pdf.write_bytes(b"%PDF fake pdf") + convert = converter_module.opendataloader_pdf.convert + calls = [] + + def failing_convert(input_path, output_dir, **kwargs): + calls.append([Path(path).stem for path in input_path]) + assert kwargs["format"] == "text" + assert kwargs["language"] == "en" + convert(input_path, output_dir, **kwargs) + if "document_1" in calls[-1]: + (Path(kwargs["image_dir"]) / "partial.png").write_bytes(b"partial") + message = "corrupt PDF" + raise RuntimeError(message) + + monkeypatch.setattr(converter_module.opendataloader_pdf, "convert", failing_convert) + converter = OpenDataLoaderConverter( + skip_failed_pdfs=True, + extract_images=True, + image_output_dir=tmp_path / "images", + output_format="text", + convert_kwargs={"language": "en"}, + ) + result = converter.run(sources=files, meta=[{"index": index} for index in range(len(files))]) + + assert calls == [ + ["document_0", "document_1", "document_2", "document_3", "document_4"], + ["document_0", "document_1"], + ["document_0"], + ["document_1"], + ["document_2", "document_3", "document_4"], + ] + assert [document.meta["file_path"] for document in result["documents"]] == [ + "good.pdf", + "another.pdf", + "fourth.pdf", + "fifth.pdf", + ] + assert [document.meta["index"] for document in result["documents"]] == [0, 2, 3, 4] + assert all(document.meta["output_format"] == "text" for document in result["documents"]) + image_paths = [Path(document.meta["file_path"]) for document in result["image_documents"]] + assert len(image_paths) == 4 + assert {path.name for path in image_paths} == {f"document_{index}_image_1.png" for index in (0, 2, 3, 4)} + assert set((tmp_path / "images").rglob("*.png")) == set(image_paths) + assert all(path.read_bytes() == b"fake image" for path in image_paths) + assert "Skipping PDF bad.pdf" in caplog.text + assert "splitting into smaller batches" in caplog.text + + +def test_converter_default_propagates_batch_failure(tmp_path, _mock_opendataloader, monkeypatch): + pdf_file = tmp_path / "bad.pdf" + pdf_file.write_bytes(b"%PDF fake pdf") + calls = [] + + def fail_convert(*, input_path, **_kwargs): + calls.append(input_path) + message = "corrupt PDF" + raise RuntimeError(message) + + monkeypatch.setattr(converter_module.opendataloader_pdf, "convert", fail_convert) + with pytest.raises(RuntimeError, match="corrupt PDF"): + OpenDataLoaderConverter().run(sources=[pdf_file]) + assert len(calls) == 1 + + +@pytest.mark.parametrize( + ("count", "bad_indices", "expected_calls"), + [(8, (), 1), (1024, (513,), 21), (5, (1, 3), 9), (4, (0, 1, 2, 3), 7), (1, (0,), 1)], +) +def test_converter_bisection_call_count(count, bad_indices, expected_calls, _mock_opendataloader, monkeypatch, caplog): + convert = converter_module.opendataloader_pdf.convert + calls = [] + sources = [ + ByteStream(data=b"%PDF fake pdf", meta={"file_path": f"upload/{index}.pdf", "index": index}) + for index in range(count) + ] + + def failing_convert(input_path, output_dir, **kwargs): + calls.append(input_path) + if any(Path(path).stem == f"document_{index}" for path in input_path for index in bad_indices): + message = "corrupt PDF" + raise subprocess.CalledProcessError(returncode=1, cmd=["java"], stderr=message) + convert(input_path, output_dir, **kwargs) + + monkeypatch.setattr(converter_module.opendataloader_pdf, "convert", failing_convert) + result = OpenDataLoaderConverter(skip_failed_pdfs=True).run(sources=sources, meta={"category": "test"}) + + assert len(calls) == expected_calls + assert [document.meta for document in result["documents"]] == [ + {"file_path": f"upload/{index}.pdf", "index": index, "category": "test", "output_format": "markdown"} + for index in range(count) + if index not in bad_indices + ] + assert all(document.content == "This is extracted PDF content" for document in result["documents"]) + assert result["image_documents"] == [] + for index in bad_indices: + assert f"Skipping PDF upload/{index}.pdf" in caplog.text + + +@pytest.mark.parametrize("count", [1, 4]) +def test_converter_all_failed_batches_leave_no_images(count, tmp_path, _mock_opendataloader, monkeypatch): + convert = converter_module.opendataloader_pdf.convert + + def fail_after_writing_output(input_path, output_dir, **kwargs): + convert(input_path, output_dir, **kwargs) + message = "corrupt PDF" + raise RuntimeError(message) + + monkeypatch.setattr(converter_module.opendataloader_pdf, "convert", fail_after_writing_output) + image_output_dir = tmp_path / "images" + converter = OpenDataLoaderConverter(skip_failed_pdfs=True, extract_images=True, image_output_dir=image_output_dir) + result = converter.run(sources=[ByteStream(data=b"%PDF fake pdf") for _ in range(count)]) + + assert result == {"documents": [], "image_documents": []} + assert not list(image_output_dir.rglob("*.png")) + + +@pytest.mark.parametrize("skip_failed_pdfs", [False, True]) +def test_converter_missing_output(skip_failed_pdfs, _mock_opendataloader, monkeypatch, caplog): + convert = converter_module.opendataloader_pdf.convert + calls = [] + + def missing_output(input_path, output_dir, **kwargs): + calls.append([Path(path).stem for path in input_path]) + convert(input_path, output_dir, **kwargs) + if "document_1" in calls[-1]: + (Path(output_dir) / "document_1.md").unlink() + + monkeypatch.setattr(converter_module.opendataloader_pdf, "convert", missing_output) + sources = [ByteStream(data=b"%PDF fake pdf") for _ in range(2)] + converter = OpenDataLoaderConverter(skip_failed_pdfs=skip_failed_pdfs) + if skip_failed_pdfs: + result = converter.run(sources=sources) + assert [document.meta["file_path"] for document in result["documents"]] == ["document_0.pdf"] + assert calls == [["document_0", "document_1"], ["document_0"], ["document_1"]] + assert "Skipping PDF document_1.pdf" in caplog.text + else: + with pytest.raises(RuntimeError, match="did not create expected file"): + converter.run(sources=sources) + assert calls == [["document_0", "document_1"]] + + def test_converter_serialization(tmp_path): image_output_dir = tmp_path / "images" converter = OpenDataLoaderConverter( @@ -436,6 +586,7 @@ def test_converter_serialization(tmp_path): convert_kwargs={"hybrid": "docling-fast"}, extract_images=True, image_output_dir=image_output_dir, + skip_failed_pdfs=True, ) data = converter.to_dict() restored = OpenDataLoaderConverter.from_dict(data) @@ -445,6 +596,7 @@ def test_converter_serialization(tmp_path): assert restored.convert_kwargs["hybrid"] == "docling-fast" assert restored.extract_images is True assert restored.image_output_dir == image_output_dir + assert restored.skip_failed_pdfs is True @pytest.mark.integration @@ -463,6 +615,29 @@ def test_real_pdf_conversion(): assert document.meta["output_format"] == "markdown" +@pytest.mark.integration +@pytest.mark.parametrize("extract_images", [False, True]) +def test_real_pdf_conversion_skips_corrupt_pdf(tmp_path, extract_images): + good_pdf = Path(__file__).parent / "test_files" / ("pdf_with_image.pdf" if extract_images else "hello_world.pdf") + bad_pdf = tmp_path / "corrupt.pdf" + bad_pdf.write_bytes(b"not a PDF") + converter = OpenDataLoaderConverter( + skip_failed_pdfs=True, + extract_images=extract_images, + image_output_dir=tmp_path / "images" if extract_images else None, + ) + + result = converter.run(sources=[good_pdf, bad_pdf]) + + assert [document.meta["file_path"] for document in result["documents"]] == [good_pdf.name] + assert result["documents"][0].content + if extract_images: + assert result["image_documents"] + assert all(Path(document.meta["file_path"]).exists() for document in result["image_documents"]) + else: + assert result["image_documents"] == [] + + @pytest.mark.integration def test_real_pdf_conversion_extracts_images(tmp_path): pdf_file = Path(__file__).parent / "test_files" / "pdf_with_image.pdf"