Coverage for src/qdrant_loader/core/chunking/docling/structure.py: 100%

39 statements  

« prev     ^ index     » next       coverage.py v7.15.0, created at 2026-07-20 10:15 +0000

1"""The anti-corruption layer: docling chunk metadata -> engine-neutral structure. 

2 

3Docling computes layout/provenance during conversion (heading paths, page numbers, 

4bounding boxes, char spans, element labels). The legacy chunker throws this away and 

5*re-derives* a thin version with regex over markdown. This module projects docling's 

6typed metadata into a small, frozen, JSON-serialisable :class:`ChunkStructure` so 

7docling classes never reach the Qdrant payload schema and the engine stays swappable 

8(the same boundary discipline the conversion engine uses). 

9 

10Docling types are confined to this module and :mod:`.docling_chunker`; everything 

11downstream (the mapper, the payload) sees only :class:`ChunkStructure`. 

12""" 

13 

14from __future__ import annotations 

15 

16from dataclasses import dataclass 

17from typing import TYPE_CHECKING, cast 

18 

19if TYPE_CHECKING: 

20 from docling_core.transforms.chunker import BaseChunk, DocMeta 

21 from docling_core.types.doc.base import BoundingBox 

22 

23 

24@dataclass(frozen=True, slots=True) 

25class BoundingBoxSpan: 

26 """An element's bounding box, projected off docling's ``BoundingBox``. 

27 

28 ``coord_origin`` is carried verbatim (``TOPLEFT`` / ``BOTTOMLEFT``) so a 

29 citation overlay can interpret the coordinates without guessing. 

30 """ 

31 

32 left: float 

33 top: float 

34 right: float 

35 bottom: float 

36 coord_origin: str 

37 

38 

39@dataclass(frozen=True, slots=True) 

40class ChunkStructure: 

41 """Engine-neutral structure/provenance for one chunk. 

42 

43 Every field maps to a docling metadata source (noted inline below). Defaults are 

44 the "absent" case (native markdown, no layout) so a chunk with no provenance is 

45 representable without ``None``-checking each field at the call site. 

46 """ 

47 

48 heading_path: tuple[str, ...] = () # meta.headings — also the embed-time prefix 

49 heading_level: int | None = None # SectionHeaderItem.level (depth) 

50 doc_items: tuple[str, ...] = () # doc_items[].label values 

51 is_table: bool = False # label rollup — element-type filter 

52 is_picture: bool = False 

53 page_start: int | None = None # min(prov.page_no) — page-scoped filter / citation 

54 page_end: int | None = None # max(prov.page_no) 

55 bbox: tuple[ 

56 BoundingBoxSpan, ... 

57 ] = () # prov.bbox — payload-only (high cardinality) 

58 charspan: tuple[int, int] | None = None # prov.charspan — source-span citation 

59 caption: str | None = None # Table/PictureItem.captions (best-effort) 

60 dl_meta_version: str | None = None # DocMeta.version — reproducibility 

61 

62 

63class StructureProjector: 

64 """Projects a docling chunk's ``meta`` into a :class:`ChunkStructure`. 

65 

66 Stateless — it holds no docling object, only the projection rules, so one 

67 instance is shared across all chunks of a document. ``project`` performs the 

68 field-by-field translation (headings, ``SectionHeaderItem.level``, the 

69 ``doc_items[].label`` rollup, the ``prov`` page/bbox/charspan aggregation). 

70 """ 

71 

72 def project(self, chunk: BaseChunk) -> ChunkStructure: 

73 # HybridChunker always yields DocChunks; narrow BaseMeta -> DocMeta so the 

74 # headings/doc_items/version fields it actually carries are visible. 

75 meta = cast("DocMeta", chunk.meta) 

76 heading_path = tuple(meta.headings or ()) 

77 doc_items = tuple(item.label.value for item in meta.doc_items) 

78 provenances = [prov for item in meta.doc_items for prov in (item.prov or ())] 

79 pages = [prov.page_no for prov in provenances if prov.page_no is not None] 

80 return ChunkStructure( 

81 heading_path=heading_path, 

82 heading_level=len(heading_path) or None, 

83 doc_items=doc_items, 

84 is_table="table" in doc_items, 

85 is_picture="picture" in doc_items, 

86 page_start=min(pages) if pages else None, 

87 page_end=max(pages) if pages else None, 

88 bbox=tuple( 

89 self._to_box(prov.bbox) for prov in provenances if prov.bbox is not None 

90 ), 

91 charspan=self._cover_spans( 

92 [prov.charspan for prov in provenances if prov.charspan is not None] 

93 ), 

94 # caption is deferred: DocMeta.captions is deprecated and not carried 

95 # through merge paths. Populate from picture/table captions if/when that 

96 # enrichment lands; until then it stays best-effort-absent. 

97 caption=None, 

98 dl_meta_version=str(meta.version) if meta.version else None, 

99 ) 

100 

101 @staticmethod 

102 def _to_box(bbox: BoundingBox) -> BoundingBoxSpan: 

103 """Project docling's ``BoundingBox`` to floats + a string coord-origin.""" 

104 return BoundingBoxSpan( 

105 left=float(bbox.l), 

106 top=float(bbox.t), 

107 right=float(bbox.r), 

108 bottom=float(bbox.b), 

109 coord_origin=bbox.coord_origin.value, 

110 ) 

111 

112 @staticmethod 

113 def _cover_spans(spans: list[tuple[int, int]]) -> tuple[int, int] | None: 

114 """The char range covering every item's span: (min start, max end).""" 

115 if not spans: 

116 return None 

117 return (min(start for start, _ in spans), max(end for _, end in spans))