Coverage for src/qdrant_loader/core/document.py: 91%
141 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-20 10:15 +0000
1import hashlib
2import uuid
3from datetime import UTC, datetime
4from typing import TYPE_CHECKING, Any
6from pydantic import BaseModel, ConfigDict, Field, PrivateAttr
8from qdrant_loader.utils.logging import LoggingConfig
10if TYPE_CHECKING:
11 from qdrant_loader.core.conversion.outcome import ConvertedDocument
13logger = LoggingConfig.get_logger(__name__)
16class Document(BaseModel):
17 """Document model with enhanced metadata support."""
19 id: str
20 title: str
21 content_type: str
22 content: str
23 contextual_content: str | None = (
24 None # Optional field for contextual embedding content
25 )
26 metadata: dict[str, Any] = Field(default_factory=dict)
27 content_hash: str
28 source_type: str
29 source: str
30 url: str
31 is_deleted: bool = False
32 created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
33 updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
35 # Transient, in-process only: the structured artifact produced by the docling
36 # conversion engine. It rides this Document by reference from the connector to the
37 # docling chunking strategy (the pipeline never serializes Documents between those
38 # stages). Held as a PrivateAttr — not a pydantic field — so it stays out of the
39 # schema, model_dump, the Qdrant payload and the content hash, and so pydantic never
40 # has to resolve docling's types (which are import-light / TYPE_CHECKING-only here).
41 _converted_document: "ConvertedDocument | None" = PrivateAttr(default=None)
43 model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid")
45 @property
46 def converted_document(self) -> "ConvertedDocument | None":
47 """The structured docling artifact for the docling chunking path, if any."""
48 return self._converted_document
50 @converted_document.setter
51 def converted_document(self, value: "ConvertedDocument | None") -> None:
52 self._converted_document = value
54 def __init__(self, **data):
55 # Generate ID only if not provided
56 if "id" not in data or not data["id"]:
57 data["id"] = self.generate_id(
58 data["source_type"], data["source"], data["url"]
59 )
61 # Calculate content hash
62 data["content_hash"] = self.calculate_content_hash(
63 data["content"], data["title"], data["metadata"]
64 )
66 # Initialize with provided data
67 super().__init__(**data)
69 # Single consolidated debug log for document creation (reduces verbosity)
70 logger.debug(
71 "Created document",
72 id=self.id,
73 content_length=len(self.content) if self.content else 0,
74 source_type=self.source_type,
75 )
77 def to_dict(self) -> dict[str, Any]:
78 """Convert document to dictionary format for Qdrant."""
79 return {
80 "id": self.id,
81 "content": self.content,
82 "contextual_content": self.contextual_content,
83 "metadata": self.metadata,
84 "source": self.source,
85 "source_type": self.source_type,
86 "created_at": self.created_at.isoformat(),
87 "updated_at": self.updated_at.isoformat(),
88 "title": self.title,
89 "url": self.url,
90 "content_hash": self.content_hash,
91 "is_deleted": self.is_deleted,
92 }
94 @classmethod
95 def from_dict(cls, data: dict[str, Any]) -> "Document":
96 """Create document from dictionary format."""
97 metadata = data.get("metadata", {})
98 doc = cls(
99 id=cls.generate_id(data["source_type"], data["source"], data["url"]),
100 content=data["content"],
101 source=data["source"],
102 source_type=data["source_type"],
103 created_at=datetime.fromisoformat(
104 data.get("created_at", datetime.now(UTC).isoformat())
105 ),
106 url=metadata.get("url"),
107 title=data["title"],
108 updated_at=metadata.get("updated_at", None),
109 content_hash=cls.calculate_content_hash(
110 data["content"], data["title"], metadata
111 ),
112 is_deleted=data.get("is_deleted", False),
113 )
114 # Add any additional metadata
115 for key, value in metadata.items():
116 if key not in [
117 "url",
118 "source",
119 "source_type",
120 "created_at",
121 "updated_at",
122 "title",
123 "content",
124 "id",
125 "content_hash",
126 ]:
127 doc.metadata[key] = value
129 return doc
131 @staticmethod
132 def calculate_content_hash(
133 content: str, title: str, metadata: dict[str, Any]
134 ) -> str:
135 """Calculate a consistent hash of document content.
137 Args:
138 content: The document content
139 title: The document title
140 metadata: The document metadata
142 Returns:
143 A consistent hash string of the content
144 """
145 import json
146 from typing import Any
148 def normalize_value(value: Any) -> Any:
149 """Normalize a value for consistent hashing."""
150 if value is None:
151 return "null"
152 if isinstance(value, str | int | float | bool):
153 return value
154 if isinstance(value, dict):
155 return {k: normalize_value(v) for k, v in sorted(value.items())}
156 if isinstance(value, list | tuple):
157 return [normalize_value(v) for v in value]
158 return str(value)
160 # Normalize all inputs
161 normalized_content = content.replace("\r\n", "\n")
162 normalized_title = title.replace("\r\n", "\n")
163 # Keys prefixed with "__" are connector-internal bookkeeping (e.g.
164 # "__ingestion_checkpoint", a pagination cursor token) rather than
165 # actual document content. They can legitimately differ between two
166 # fetches of the very same, unchanged document, so including them
167 # would make change detection think the document was modified on
168 # every single ingestion run.
169 hashable_metadata = {
170 k: v for k, v in metadata.items() if not k.startswith("__")
171 }
172 normalized_metadata = normalize_value(hashable_metadata)
174 # Create a consistent string representation
175 content_string = json.dumps(
176 {
177 "content": normalized_content,
178 "title": normalized_title,
179 "metadata": normalized_metadata,
180 },
181 sort_keys=True,
182 ensure_ascii=False,
183 )
185 # Generate SHA-256 hash
186 content_hash = hashlib.sha256(content_string.encode("utf-8")).hexdigest()
188 return content_hash
190 @staticmethod
191 def generate_id(source_type: str, source: str, url: str) -> str:
192 """Generate a consistent document ID based on source attributes.
194 Args:
195 source_type: The type of source (e.g., 'publicdocs', 'confluence', etc.)
196 source: The source identifier
197 url: Optional URL of the document
199 Returns:
200 A consistent UUID string generated from the inputs
201 """
202 from urllib.parse import urlparse, urlunparse
204 logger = LoggingConfig.get_logger(__name__)
206 def normalize_url(url: str) -> str:
207 """Normalize a URL for consistent hashing.
209 This function normalizes URLs by:
210 1. Converting to lowercase
211 2. Removing trailing slashes
212 3. Removing query parameters
213 4. Removing fragments
214 5. Handling empty paths
215 6. Handling malformed URLs
216 """
217 try:
218 # Convert to lowercase first to handle case variations
219 url = url.lower().strip()
221 # Parse the URL
222 parsed = urlparse(url)
224 # Normalize the scheme and netloc (already lowercase from above)
225 scheme = parsed.scheme
226 netloc = parsed.netloc
228 # Normalize the path
229 path = parsed.path.rstrip("/")
230 if not path: # Handle empty paths
231 path = "/"
233 # Construct normalized URL without query parameters and fragments
234 normalized = urlunparse(
235 (scheme, netloc, path, "", "", "") # params # query # fragment
236 )
238 logger.debug(f"Normalized URL: {normalized}")
239 return normalized
240 except Exception as e:
241 logger.error(f"Error normalizing URL {url}: {str(e)}")
242 # If URL parsing fails, return the original URL in lowercase
243 return url.lower().strip()
245 def normalize_string(s: str) -> str:
246 """Normalize a string for consistent hashing."""
247 normalized = s.strip().lower()
248 logger.debug(f"Normalized string '{s}' to '{normalized}'")
249 return normalized
251 # Normalize all inputs
252 normalized_source_type = normalize_string(source_type)
253 normalized_source = normalize_string(source)
254 normalized_url = normalize_url(url)
256 # Create a consistent string combining all identifying elements
257 identifier = f"{normalized_source_type}:{normalized_source}:{normalized_url}"
258 logger.debug(f"Generated identifier: {identifier}")
260 # Generate a SHA-256 hash of the identifier
261 sha256_hash = hashlib.sha256(identifier.encode("utf-8")).digest()
263 # Convert the first 16 bytes to a UUID (UUID is 16 bytes)
264 # This ensures a valid UUID that Qdrant will accept
265 consistent_uuid = uuid.UUID(bytes=sha256_hash[:16])
266 logger.debug(f"Generated UUID: {consistent_uuid}")
268 return str(consistent_uuid)
270 @staticmethod
271 def generate_chunk_id(document_id: str, chunk_index: int) -> str:
272 """Generate a unique ID for a document chunk.
274 Args:
275 document_id: The parent document's ID
276 chunk_index: The index of the chunk
278 Returns:
279 A unique chunk ID
280 """
281 # Create a string combining document ID and chunk index
282 chunk_string = f"{document_id}_{chunk_index}"
284 # Hash the string to get a consistent length ID
285 chunk_hash = hashlib.sha256(chunk_string.encode()).hexdigest()
287 # Convert to UUID format for Qdrant compatibility
288 chunk_uuid = uuid.UUID(chunk_hash[:32])
290 return str(chunk_uuid)
292 # Hierarchy convenience methods
293 def get_parent_id(self) -> str | None:
294 """Get the parent document ID if available.
296 Returns:
297 Parent document ID or None if this is a root document
298 """
299 return self.metadata.get("parent_id")
301 def get_parent_title(self) -> str | None:
302 """Get the parent document title if available.
304 Returns:
305 Parent document title or None if this is a root document
306 """
307 return self.metadata.get("parent_title")
309 def get_breadcrumb(self) -> list[str]:
310 """Get the breadcrumb trail for this document.
312 Returns:
313 List of ancestor titles leading to this document
314 """
315 return self.metadata.get("breadcrumb", [])
317 def get_breadcrumb_text(self) -> str:
318 """Get the breadcrumb trail as a formatted string.
320 Returns:
321 Breadcrumb trail formatted as "Parent > Child > Current"
322 """
323 return self.metadata.get("breadcrumb_text", "")
325 def get_depth(self) -> int:
326 """Get the depth of this document in the hierarchy.
328 Returns:
329 Depth level (0 for root documents, 1 for first level children, etc.)
330 """
331 return self.metadata.get("depth", 0)
333 def get_ancestors(self) -> list[dict]:
334 """Get the list of ancestor documents.
336 Returns:
337 List of ancestor document information (id, title, type)
338 """
339 return self.metadata.get("ancestors", [])
341 def get_children(self) -> list[dict]:
342 """Get the list of child documents.
344 Returns:
345 List of child document information (id, title, type)
346 """
347 return self.metadata.get("children", [])
349 def is_root_document(self) -> bool:
350 """Check if this is a root document (no parent).
352 Returns:
353 True if this is a root document, False otherwise
354 """
355 return self.get_parent_id() is None
357 def has_children(self) -> bool:
358 """Check if this document has child documents.
360 Returns:
361 True if this document has children, False otherwise
362 """
363 return len(self.get_children()) > 0
365 def get_hierarchy_context(self) -> str:
366 """Get a formatted string describing the document's position in the hierarchy.
368 Returns:
369 Formatted hierarchy context string
370 """
371 breadcrumb = self.get_breadcrumb_text()
372 depth = self.get_depth()
373 children_count = len(self.get_children())
375 context_parts = []
377 if breadcrumb:
378 context_parts.append(f"Path: {breadcrumb}")
380 context_parts.append(f"Depth: {depth}")
382 if children_count > 0:
383 context_parts.append(f"Children: {children_count}")
385 return " | ".join(context_parts)
387 def build_contextual_content(self) -> str | None:
388 """Build a contextual prefix like:
389 [Source: confluence | Document: My Title | Project: X]\n\n
391 Returns:
392 Contextual prefix string or None if required fields are missing.
393 """
394 if not self.source or not self.title:
395 return None
396 parts = [
397 f"Source: {self.source_type}",
398 f"Title: {self.title}",
399 ]
400 project = self.metadata.get("project_name")
401 if project:
402 parts.append(f"Project: {project}")
403 return f"[{' | '.join(parts)}]\n\n"