fix location of chunks.jsonl when using the file storage target processors

Signed-off-by: Michele Dolfi <dol@zurich.ibm.com>

Author: dolfim-ibm

docling_jobkit/connectors/opensearch/target_processor.py

@@ -1,6 +1,6 @@
 import json
 from pathlib import Path
-from typing import Any, Union
+from typing import Any, Optional, Union
 
 from opensearchpy import OpenSearch
 
@@ -85,7 +85,12 @@ def instance_requires_chunks(self) -> bool:
         """True when this processor is configured as an opensearch_chunks target."""
         return isinstance(self._target, OpenSearchChunkTarget)
 
-    def begin_chunks(self, filename: str, temp_dir: Path) -> None:
+    def begin_chunks(
+        self,
+        filename: str,
+        temp_dir: Path,
+        chunk_target_key: Optional[str] = None,
+    ) -> None:
         """No file needed — each chunk is upserted directly in consume_chunk()."""
 
     def consume_chunk(self, chunk: ChunkedDocumentResultItem) -> None:

docling_jobkit/connectors/target_processor.py

@@ -118,17 +118,32 @@ def requires_chunks(cls) -> bool:
         """
         return False
 
-    def begin_chunks(self, filename: str, temp_dir: Path) -> None:
+    def begin_chunks(
+        self,
+        filename: str,
+        temp_dir: Path,
+        chunk_target_key: Optional[str] = None,
+    ) -> None:
         """Called once before the first chunk of a document is streamed.
 
         Default: open a temp ``{stem}.chunks.jsonl`` file for writing.
         DB processors that override ``requires_chunks()`` can use this to
         initialise per-document state instead.
+
+        ``chunk_target_key``, when provided, is the storage path that
+        ``end_chunks`` will pass to ``upload_file()`` as *target_filename*.
+        When omitted the bare filename (``{stem}.chunks.jsonl``) is used,
+        which is only correct for database processors that override
+        ``end_chunks`` and never call ``upload_file()`` themselves.
         """
-        self._chunk_jsonl_path: Optional[Path] = (
-            temp_dir / f"{Path(filename).stem}.chunks.jsonl"
-        )
+        stem = Path(filename).stem
+        self._chunk_jsonl_path: Optional[Path] = temp_dir / f"{stem}.chunks.jsonl"
         self._chunk_jsonl_file = self._chunk_jsonl_path.open("w", encoding="utf-8")
+        self._chunk_target_key: Optional[str] = (
+            chunk_target_key
+            if chunk_target_key is not None
+            else self._chunk_jsonl_path.name
+        )
 
     def consume_chunk(self, chunk: "ChunkedDocumentResultItem") -> None:
         """Called once per chunk as it is produced by the chunker.
@@ -154,6 +169,6 @@ def end_chunks(self) -> None:
         self._chunk_jsonl_file.close()
         self.upload_file(
             filename=self._chunk_jsonl_path,  # type: ignore[arg-type]
-            target_filename=self._chunk_jsonl_path.name,  # type: ignore[union-attr]
+            target_filename=self._chunk_target_key,  # type: ignore[arg-type]
             content_type="application/jsonl",
         )

docling_jobkit/convert/export.py

@@ -71,6 +71,7 @@ def stream_chunks_for_document(
     processors: list[BaseTargetProcessor],
     chunks_in_formats: bool,
     temp_dir: Path,
+    chunk_target_key: str | None = None,
 ) -> int:
     """Run the streaming chunk protocol across *processors* for one document.
 
@@ -90,6 +91,10 @@ def stream_chunks_for_document(
     - DB processors that override ``requires_chunks()`` always participate.
     - File-storage processors participate only when *chunks_in_formats* is
       True, and only if they are *not* database processors.
+
+    ``chunk_target_key`` is forwarded to ``begin_chunks`` so that file-storage
+    processors write the chunks file under the correct subdirectory path (e.g.
+    ``chunks/bo20/1016445.chunks.jsonl``).  DB processors ignore it.
     """
     if not _is_exportable_status(exportable_document.status):
         return 0
@@ -106,7 +111,7 @@ def stream_chunks_for_document(
         return 0
 
     for p in chunk_processors:
-        p.begin_chunks(filename, temp_dir)
+        p.begin_chunks(filename, temp_dir, chunk_target_key=chunk_target_key)
     n = 0
     try:
         for chunk in chunker_manager.chunk_document(

docling_jobkit/convert/results.py

@@ -650,6 +650,8 @@ def _iter_remote_documents(
             num_failed += 1
 
         if chunk_active:
+            _source_identity = _resolve_source_identity(task, exportable_document, idx)
+            _stem = exportable_document.file.stem
             with tempfile.TemporaryDirectory(dir=work_dir) as _chunk_tmp:
                 stream_chunks_for_document(
                     exportable_document=final_document,
@@ -659,6 +661,7 @@ def _iter_remote_documents(
                     processors=processors,
                     chunks_in_formats=chunks_in_formats,
                     temp_dir=Path(_chunk_tmp),
+                    chunk_target_key=f"{_source_identity.source_key}/{_stem}.chunks.jsonl",
                 )
 
     return processed_docs, num_succeeded, num_partially_succeeded, num_failed

docling_jobkit/convert/results_processor.py

@@ -284,6 +284,9 @@ def process_documents(self, results: Iterable[ConversionResult]):
                                     processors=processors,
                                     chunks_in_formats=self._chunks_in_formats,
                                     temp_dir=temp_dir,
+                                    chunk_target_key=self._target_key(
+                                        f"chunks/{name_without_ext}.chunks.jsonl"
+                                    ),
                                 )
 
                             yield f"{doc_hash} - SUCCESS"