Coverage for src/qdrant_loader/connectors/localfile/connector.py: 91%
121 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-20 10:15 +0000
1import os
2from collections.abc import AsyncIterator
3from datetime import UTC, datetime
4from urllib.parse import unquote, urlparse
6from qdrant_loader.connectors.base import BaseConnector, resolve_safe_path
7from qdrant_loader.core.conversion.service import ConversionService
8from qdrant_loader.core.document import Document
9from qdrant_loader.core.file_conversion import (
10 FileConversionConfig,
11 FileDetector,
12)
13from qdrant_loader.utils.logging import LoggingConfig
14from qdrant_loader.utils.sensitive import sanitize_exception_message
16from .config import LocalFileConfig
17from .file_processor import LocalFileFileProcessor
18from .metadata_extractor import LocalFileMetadataExtractor
21class LocalFileConnector(BaseConnector):
22 """Connector for ingesting local files."""
24 def __init__(self, config: LocalFileConfig):
25 super().__init__(config)
26 self.config = config
27 # Parse base_url (file://...) to get the local path with Windows support
28 parsed = urlparse(str(config.base_url))
29 self.base_path = self._fix_windows_file_path(parsed.path)
30 self.file_processor = LocalFileFileProcessor(config, self.base_path)
31 self.metadata_extractor = LocalFileMetadataExtractor(self.base_path)
32 self.logger = LoggingConfig.get_logger(__name__)
33 self._initialized = True
35 # Initialize file conversion components if enabled
36 self.conversion_service = None
37 self.file_detector = None
38 if self.config.enable_file_conversion:
39 self.logger.debug("File conversion enabled for LocalFile connector")
40 # File conversion config will be set from global config during ingestion
41 self.file_detector = FileDetector()
42 # Update file processor with file detector
43 self.file_processor = LocalFileFileProcessor(
44 config, self.base_path, self.file_detector
45 )
46 else:
47 self.logger.debug("File conversion disabled for LocalFile connector")
49 def _fix_windows_file_path(self, path: str) -> str:
50 """Fix Windows file path from URL parsing.
52 urlparse() adds a leading slash to Windows drive letters, e.g.:
53 file:///C:/Users/... -> path = "/C:/Users/..."
54 This method removes the leading slash for Windows paths and handles URL decoding.
56 Args:
57 path: Raw path from urlparse()
59 Returns:
60 Fixed path suitable for the current platform
61 """
62 # First decode URL encoding (e.g., %20 -> space)
63 path = unquote(path)
65 # Handle Windows paths: remove leading slash if it's a drive letter
66 if len(path) >= 3 and path[0] == "/" and path[2] == ":":
67 # This looks like a Windows path with leading slash: "/C:/..." or "/C:" -> "C:/..." or "C:"
68 path = path[1:]
70 return path
72 def set_file_conversion_config(self, file_conversion_config: FileConversionConfig):
73 """Set file conversion configuration from global config.
75 Args:
76 file_conversion_config: Global file conversion configuration
77 """
78 if self.config.enable_file_conversion:
79 self.conversion_service = ConversionService(file_conversion_config)
80 self.logger.debug(
81 "Conversion service initialized",
82 engine=str(file_conversion_config.engine),
83 )
85 def _build_file_document(self, file_path: str) -> Document | None:
86 """Build a Document for a local file, or None if the file should be skipped.
88 Used by both stream_documents and fetch_by_id.
89 """
90 file = os.path.basename(file_path)
91 # Get relative path from base directory
92 rel_path = os.path.relpath(file_path, self.base_path)
93 file_extension = os.path.splitext(file)[1].lower()
95 if self.config.enable_file_conversion and file_extension in {".doc", ".ppt"}:
96 file_info = (
97 self.file_detector.get_file_type_info(file_path)
98 if self.file_detector
99 else {
100 "mime_type": None,
101 "file_extension": file_extension,
102 }
103 )
104 self.logger.warning(
105 "Skipping file: old doc/ppt are not supported for MarkItDown conversion",
106 file_path=rel_path.replace("\\", "/"),
107 mime_type=file_info.get("mime_type"),
108 file_extension=file_info.get("file_extension"),
109 )
110 return None
112 # Check if file needs conversion. The eligibility gate is the
113 # ConversionService (engine-aware): for docling it also enforces
114 # the enabled-format + size policy, so a file the active engine
115 # can't convert is skipped here instead of falling into a fallback.
116 needs_conversion = (
117 self.config.enable_file_conversion
118 and self.conversion_service is not None
119 and self.conversion_service.is_supported(file_path)
120 )
122 converted_document = None
123 original_file_type = None
124 if needs_conversion:
125 self.logger.debug(
126 "File needs conversion",
127 file_path=rel_path.replace("\\", "/"),
128 )
129 # ConversionService picks the engine (markitdown|docling) and
130 # returns success or a fallback document — it never raises for a
131 # per-document conversion failure.
132 assert self.conversion_service is not None # Type checker hint
133 converted = self.conversion_service.convert(file_path)
134 content = converted.content
135 content_type = "md" # Converted files (and fallbacks) are markdown
136 conversion_method = converted.conversion_method
137 conversion_failed = converted.conversion_failed
138 converted_document = converted.converted_document
139 original_file_type = converted.original_file_type
140 self.logger.info(
141 "File conversion completed",
142 file_path=rel_path.replace("\\", "/"),
143 conversion_method=conversion_method,
144 conversion_failed=conversion_failed,
145 )
146 else:
147 # Read file content normally
148 with open(file_path, encoding="utf-8", errors="ignore") as f:
149 content = f.read()
150 # Get file extension without the dot
151 content_type = os.path.splitext(file)[1].lower().lstrip(".")
152 conversion_method = None
153 conversion_failed = False
155 # Get file modification time
156 file_mtime = os.path.getmtime(file_path)
157 updated_at = datetime.fromtimestamp(file_mtime, tz=UTC)
159 metadata = self.metadata_extractor.extract_all_metadata(file_path, content)
161 # Add file conversion metadata if applicable
162 if needs_conversion:
163 metadata.update(
164 {
165 "conversion_method": conversion_method,
166 "conversion_failed": conversion_failed,
167 "original_file_type": original_file_type,
168 }
169 )
171 self.logger.debug(f"Processed local file: {rel_path.replace('\\', '/')}")
173 # Create consistent URL with forward slashes for cross-platform compatibility
174 normalized_path = os.path.realpath(file_path).replace("\\", "/")
175 doc = Document(
176 title=os.path.basename(file_path),
177 content=content,
178 content_type=content_type,
179 metadata=metadata,
180 source_type="localfile",
181 source=self.config.source,
182 url=f"file://{normalized_path}",
183 is_deleted=False,
184 updated_at=updated_at,
185 )
186 # Carry the structured artifact (docling path) to the chunker,
187 # in-process and by reference — never serialized.
188 doc.converted_document = converted_document
189 return doc
191 def _is_within_base_path(self, file_path: str) -> bool:
192 base_real = os.path.realpath(self.base_path)
193 file_real = os.path.realpath(file_path)
194 return file_real == base_real or file_real.startswith(base_real + os.sep)
196 async def stream_documents(
197 self, since: datetime | None = None
198 ) -> AsyncIterator[Document]:
199 """Stream documents from the local file source (WS-1 connector contract).
201 Note:
202 The `since` parameter is not yet implemented for incremental
203 ingestion. All files are processed regardless of modification time.
204 """
205 for root, _, files in os.walk(self.base_path):
206 for file in files:
207 file_path = os.path.join(root, file)
208 if not self._is_within_base_path(file_path):
209 self.logger.warning(
210 "Skipping file outside base path",
211 file_path=file_path.replace("\\", "/"),
212 )
213 continue
214 if not self.file_processor.should_process_file(file_path):
215 continue
216 try:
217 doc = self._build_file_document(file_path)
218 if doc is not None:
219 yield doc
220 except (OSError, UnicodeError, ValueError) as e:
221 self.logger.warning(
222 "Failed to process file",
223 file_path=file_path.replace("\\", "/"),
224 error=sanitize_exception_message(e),
225 )
226 continue
227 except Exception:
228 raise
230 async def get_documents(self) -> list[Document]:
231 """Get documents from the local file source (DEPRECATED - use stream_documents)."""
232 return await super().get_documents()
234 async def fetch_by_id(self, entity_id: str) -> Document | None:
235 """Fetch a single file by its path relative to the connector's base directory."""
236 file_path = resolve_safe_path(self.base_path, entity_id)
237 if file_path is None:
238 self.logger.warning("Path traversal attempt blocked", entity_id=entity_id)
239 return None
240 if not os.path.exists(file_path) or not self.file_processor.should_process_file(
241 file_path
242 ):
243 return None
244 try:
245 return self._build_file_document(file_path)
246 except (OSError, UnicodeError, ValueError) as e:
247 self.logger.warning(
248 "Failed to fetch file by id",
249 entity_id=entity_id,
250 error=sanitize_exception_message(e),
251 )
252 return None
254 async def list_entity_ids(self) -> AsyncIterator[str]:
255 """Stream paths (relative to base_path, forward slashes) for all processable files."""
256 for root, _, files in os.walk(self.base_path):
257 for file in files:
258 file_path = os.path.join(root, file)
259 if not self._is_within_base_path(file_path):
260 continue
261 if self.file_processor.should_process_file(file_path):
262 rel_path = os.path.relpath(file_path, self.base_path)
263 yield rel_path.replace(os.sep, "/").replace("\\", "/")