Coverage for src/qdrant_loader/core/chunking/strategy/markdown/chunk_processor.py: 100%

1"""Chunk processing coordination for markdown strategy.""" 

2 

3from typing import TYPE_CHECKING, Any 

4 

5import structlog 

6 

7from qdrant_loader.core.document import Document 

8from qdrant_loader.core.text_processing.chunk_enricher import ChunkEnricher 

9 

10if TYPE_CHECKING: 

11 from qdrant_loader.config import Settings 

12 

13logger = structlog.get_logger(__name__) 

14 

15 

16class ChunkProcessor: 

17 """Handles chunk processing coordination including parallel execution and semantic analysis.""" 

18 

19 def __init__(self, settings: "Settings"): 

20 """Initialize the chunk processor. 

21 

22 Args: 

23 settings: Configuration settings 

24 """ 

25 self.settings = settings 

26 

27 # The shared enricher owns the SemanticAnalyzer (and the enable/enhanced 

28 # gating); markdown and docling enrich through the same code path. 

29 self._enricher = ChunkEnricher(settings) 

30 

31 # Cache for processed chunks to avoid recomputation 

32 self._processed_chunks: dict[str, dict[str, Any]] = {} 

33 

34 def process_chunk( 

35 self, chunk: str, chunk_index: int, total_chunks: int 

36 ) -> dict[str, Any]: 

37 """Process a single chunk's semantic enrichment. 

38 

39 Args: 

40 chunk: The chunk to process 

41 chunk_index: Index of the chunk 

42 total_chunks: Total number of chunks 

43 

44 Returns: 

45 Dictionary containing enrichment results 

46 """ 

47 logger.debug( 

48 "Processing chunk", 

49 chunk_index=chunk_index, 

50 total_chunks=total_chunks, 

51 chunk_length=len(chunk), 

52 ) 

53 results = self._enricher.enrich(chunk, doc_id=f"chunk_{chunk_index}") 

54 self._processed_chunks[chunk] = results 

55 return results 

56 

57 def create_chunk_document( 

58 self, 

59 original_doc: Document, 

60 chunk_content: str, 

61 chunk_index: int, 

62 total_chunks: int, 

63 chunk_metadata: dict[str, Any], 

64 skip_nlp: bool = False, 

65 ) -> Document: 

66 """Create a chunk document with enhanced metadata. 

67 

68 Args: 

69 original_doc: Original document being chunked 

70 chunk_content: Content of the chunk 

71 chunk_index: Index of the chunk 

72 total_chunks: Total number of chunks 

73 chunk_metadata: Chunk-specific metadata 

74 skip_nlp: Whether to skip NLP processing 

75 

76 Returns: 

77 Document representing the chunk 

78 """ 

79 # Create base chunk document 

80 chunk_doc = Document( 

81 content=chunk_content, 

82 title=f"{original_doc.title} - Chunk {chunk_index + 1}", 

83 source=original_doc.source, 

84 source_type=original_doc.source_type, 

85 url=original_doc.url, 

86 content_type=original_doc.content_type, 

87 metadata=original_doc.metadata.copy(), 

88 ) 

89 

90 # 🔥 FIX: Manually assign chunk ID (following pattern from other strategies) 

91 chunk_doc.id = Document.generate_chunk_id(original_doc.id, chunk_index) 

92 

93 # Add chunk-specific metadata 

94 chunk_doc.metadata.update(chunk_metadata) 

95 chunk_doc.metadata.update( 

96 { 

97 "chunk_index": chunk_index, 

98 "total_chunks": total_chunks, 

99 "chunk_size": len(chunk_content), 

100 "parent_document_id": original_doc.id, 

101 "chunking_strategy": "markdown", 

102 } 

103 ) 

104 

105 # Perform semantic analysis if not skipped 

106 if not skip_nlp: 

107 semantic_results = self.process_chunk( 

108 chunk_content, chunk_index, total_chunks 

109 ) 

110 chunk_doc.metadata.update(semantic_results) 

111 

112 return chunk_doc 

113 

114 def estimate_chunk_count(self, content: str) -> int: 

115 """Estimate the number of chunks that will be generated. 

116 

117 Args: 

118 content: The content to estimate chunks for 

119 

120 Returns: 

121 int: Estimated number of chunks 

122 """ 

123 chunk_size = self.settings.global_config.chunking.chunk_size 

124 

125 # Simple estimation: total chars / chunk_size 

126 # This is approximate since we split by paragraphs and have overlap 

127 estimated = len(content) // chunk_size 

128 

129 # Add some buffer for overlap and paragraph boundaries 

130 # Apply estimation buffer from configuration 

131 buffer_factor = ( 

132 1.0 

133 + self.settings.global_config.chunking.strategies.markdown.estimation_buffer 

134 ) 

135 estimated = int(estimated * buffer_factor) 

136 

137 return max(1, estimated) # At least 1 chunk 

138 

139 def shutdown(self): 

140 """Shutdown and clean up resources.""" 

141 if getattr(self, "_enricher", None) is not None: 

142 self._enricher.shutdown() 

143 

144 def __del__(self): 

145 """Cleanup on deletion.""" 

146 self.shutdown()