Coverage for src/qdrant_loader/core/conversion/outcome.py: 95%

1"""The conversion -> chunking handoff types, and the typed failure model. 

2 

3Conversion produces a *structured* artifact — a ``DoclingDocument`` wrapped in 

4:class:`ConvertedDocument` — or a typed :class:`ConversionOutcome`. It never 

5returns a fake markdown stub carrying an error string: failure is a value the caller 

6inspects, not content that flows into embedding. 

7 

8Per the conversion/chunking boundary, the structured document is the contract; 

9``to_markdown`` is a convenience for the prose path, and the chunker chooses its 

10own view (structured table traversal vs. markdown) per format. 

11""" 

12 

13from __future__ import annotations 

14 

15from dataclasses import dataclass 

16from enum import StrEnum 

17from typing import TYPE_CHECKING 

18 

19if TYPE_CHECKING: 

20 from docling.datamodel.document import ConversionResult 

21 from docling_core.types.doc import DoclingDocument 

22 

23 

24class ConversionStatus(StrEnum): 

25 SUCCESS = "success" 

26 PARTIAL = "partial" # docling PARTIAL_SUCCESS (e.g. document_timeout mid-doc) 

27 FAILED = "failed" 

28 SKIPPED = "skipped" # rejected by policy (format not enabled / too large) 

29 

30 

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

32class ConvertedDocument: 

33 """The canonical conversion artifact handed to the chunking layer. 

34 

35 Holds the structured ``DoclingDocument`` so consumers can read cells/tables 

36 directly — no serialize-then-reparse round-trip. 

37 """ 

38 

39 document: DoclingDocument 

40 source_format: str # the InputFormat value, e.g. "pdf" 

41 

42 def to_markdown(self) -> str: 

43 """One-way serialization for the prose path. Not the primary contract.""" 

44 return self.document.export_to_markdown() 

45 

46 

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

48class ConversionOutcome: 

49 """The result value of a conversion attempt; the caller owns skip-vs-stub policy.""" 

50 

51 status: ConversionStatus 

52 document: ConvertedDocument | None = None 

53 error: str | None = None 

54 

55 @property 

56 def succeeded(self) -> bool: 

57 return self.status in (ConversionStatus.SUCCESS, ConversionStatus.PARTIAL) 

58 

59 @classmethod 

60 def skipped(cls, reason: str) -> ConversionOutcome: 

61 """A source the policy declined to convert (format/size).""" 

62 return cls(status=ConversionStatus.SKIPPED, error=reason) 

63 

64 @classmethod 

65 def failed(cls, error: str) -> ConversionOutcome: 

66 """The engine attempted conversion and failed.""" 

67 return cls(status=ConversionStatus.FAILED, error=error) 

68 

69 @classmethod 

70 def from_docling_result(cls, docling_result: ConversionResult) -> ConversionOutcome: 

71 """Translate a docling ``ConversionResult`` into our typed outcome.""" 

72 from docling.datamodel.base_models import ConversionStatus as DoclingStatus 

73 

74 if docling_result.status in ( 

75 DoclingStatus.SUCCESS, 

76 DoclingStatus.PARTIAL_SUCCESS, 

77 ): 

78 status = ( 

79 ConversionStatus.SUCCESS 

80 if docling_result.status is DoclingStatus.SUCCESS 

81 else ConversionStatus.PARTIAL 

82 ) 

83 source_format = docling_result.input.format.value 

84 # Repair docling's XLSX defects (merged-cell duplication, dropped 

85 # sheet names) once on the structured document, before either the 

86 # markdown export or the chunker reads it. No-op for other formats. 

87 from .postprocess import postprocess_spreadsheet 

88 

89 postprocess_spreadsheet(docling_result.document, source_format) 

90 document = ConvertedDocument( 

91 document=docling_result.document, 

92 source_format=source_format, 

93 ) 

94 return cls(status=status, document=document) 

95 

96 message = ( 

97 "; ".join(error.error_message for error in docling_result.errors) 

98 or "conversion failed" 

99 ) 

100 return cls(status=ConversionStatus.FAILED, error=message)