Coverage for src/qdrant_loader/connectors/publicdocs/connector.py: 81%
318 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-20 10:15 +0000
1"""Public documentation connector implementation."""
3import fnmatch
4import logging
5import warnings
6from collections import deque
7from collections.abc import AsyncIterator
8from datetime import UTC, datetime
9from typing import cast
10from urllib.parse import urljoin, urlparse
12import aiohttp
13from bs4 import BeautifulSoup, XMLParsedAsHTMLWarning
15from qdrant_loader.connectors.base import BaseConnector
16from qdrant_loader.connectors.exceptions import (
17 ConnectorError,
18 ConnectorNotInitializedError,
19 DocumentProcessingError,
20 HTTPRequestError,
21)
22from qdrant_loader.connectors.publicdocs.config import PublicDocsSourceConfig
23from qdrant_loader.connectors.publicdocs.crawler import (
24 discover_pages as _discover_pages,
25)
27# Local HTTP helper for safe text reading
28from qdrant_loader.connectors.publicdocs.http import read_text_response as _read_text
29from qdrant_loader.connectors.shared.http import (
30 RateLimiter,
31)
32from qdrant_loader.connectors.shared.http import (
33 aiohttp_request_with_policy as _aiohttp_request,
34)
35from qdrant_loader.core.attachment_downloader import (
36 AttachmentDownloader,
37 AttachmentMetadata,
38)
39from qdrant_loader.core.document import Document
40from qdrant_loader.core.file_conversion import (
41 FileConversionConfig,
42 FileConverter,
43 FileDetector,
44)
45from qdrant_loader.utils.logging import LoggingConfig
47# Suppress XML parsing warning
48warnings.filterwarnings("ignore", category=XMLParsedAsHTMLWarning)
51logger = LoggingConfig.get_logger(__name__)
54class PublicDocsConnector(BaseConnector):
55 """Connector for public documentation sources."""
57 def __init__(self, config: PublicDocsSourceConfig):
58 """Initialize the connector.
60 Args:
61 config: Configuration for the public documentation source
62 state_manager: State manager for tracking document states
63 """
64 super().__init__(config)
65 self.config = config
66 self.logger = LoggingConfig.get_logger(__name__)
67 self._initialized = False
68 self.base_url = str(config.base_url)
69 self.url_queue = deque()
70 self.visited_urls = set()
71 self.version = config.version
72 self.logger.debug(
73 "Initialized PublicDocsConnector",
74 base_url=self.base_url,
75 version=self.version,
76 exclude_paths=config.exclude_paths,
77 path_pattern=config.path_pattern,
78 )
80 # Initialize file conversion components if enabled
81 self.file_converter: FileConverter | None = None
82 self.file_detector: FileDetector | None = None
83 self.attachment_downloader: AttachmentDownloader | None = None
85 if config.enable_file_conversion:
86 self.file_detector = FileDetector()
87 # FileConverter will be initialized when file_conversion_config is set
89 async def __aenter__(self):
90 """Async context manager entry."""
91 if not self._initialized:
92 self._client = aiohttp.ClientSession()
93 self._initialized = True
95 # Initialize attachment downloader with aiohttp session if needed
96 if self.config.download_attachments:
97 # Convert aiohttp session to requests session for compatibility
98 import requests
100 session = requests.Session()
101 self.attachment_downloader = AttachmentDownloader(session=session)
103 # Initialize rate limiter for crawling (configurable)
104 self._rate_limiter = RateLimiter.per_minute(self.config.requests_per_minute)
106 return self
108 async def __aexit__(self, exc_type, exc_val, _exc_tb):
109 """Async context manager exit."""
110 if self._initialized and self._client:
111 await self._client.close()
112 self._client = None
113 self._initialized = False
115 @property
116 def client(self) -> aiohttp.ClientSession:
117 """Get the client session."""
118 if not self._client or not self._initialized:
119 raise RuntimeError(
120 "Client session not initialized. Use async context manager."
121 )
122 return self._client
124 def set_file_conversion_config(self, config: FileConversionConfig) -> None:
125 """Set the file conversion configuration.
127 Args:
128 config: File conversion configuration
129 """
130 if self.config.enable_file_conversion and self.file_detector:
131 self.file_converter = FileConverter(config)
132 if self.config.download_attachments and self.attachment_downloader:
133 # Reinitialize attachment downloader with file conversion config
134 import requests
136 session = requests.Session()
137 self.attachment_downloader = AttachmentDownloader(
138 session=session,
139 file_conversion_config=config,
140 enable_file_conversion=True,
141 max_attachment_size=config.max_file_size,
142 )
144 def _should_process_url(self, url: str) -> bool:
145 """Check if a URL should be processed based on configuration."""
146 self.logger.debug(f"Checking if URL should be processed: {url}")
148 # Check if URL matches base URL
149 if not url.startswith(str(self.base_url)):
150 self.logger.debug(f"URL does not match base URL: {url}")
151 return False
152 self.logger.debug(f"URL matches base URL: {url}")
154 # Extract path from URL
155 path = url[len(str(self.base_url)) :]
156 self.logger.debug(f"Extracted path from URL: {path}")
158 # Check exclude paths
159 for exclude_path in self.config.exclude_paths:
160 self.logger.debug(f"Checking exclude path: {exclude_path} against {path}")
161 if fnmatch.fnmatch(path, exclude_path):
162 self.logger.debug(f"URL path matches exclude pattern: {path}")
163 return False
164 self.logger.debug(f"URL path not in exclude paths: {path}")
166 # Check path pattern
167 if self.config.path_pattern is None:
168 self.logger.debug("No path pattern specified, skipping pattern check")
169 return True
171 self.logger.debug(f"Checking path pattern: {self.config.path_pattern}")
172 if not fnmatch.fnmatch(path, self.config.path_pattern):
173 self.logger.debug(f"URL path does not match pattern: {path}")
174 return False
175 self.logger.debug(f"URL path matches pattern: {path}")
177 self.logger.debug(f"URL passed all checks, will be processed: {url}")
178 return True
180 def _build_page_document(self, page: str, content: str, title: str) -> Document:
181 """Build a Document for a documentation page (used by stream_documents and fetch_by_id)."""
182 return Document(
183 title=title,
184 content=content,
185 content_type="html",
186 metadata={
187 "title": title,
188 "url": page,
189 "version": self.version,
190 },
191 source_type=self.config.source_type,
192 source=self.config.source,
193 url=page,
194 # For public docs, we don't have a created or updated date. So we use a very old date.
195 # The content hash will be the same for the same page, so it will be update if the hash changes.
196 created_at=datetime(1970, 1, 1, 0, 0, 0, 0, UTC),
197 updated_at=datetime(1970, 1, 1, 0, 0, 0, 0, UTC),
198 )
200 async def stream_documents(
201 self, since: datetime | None = None
202 ) -> AsyncIterator[Document]:
203 """Stream documentation pages from the source (WS-1 connector contract).
205 Yields:
206 Document objects from the source.
208 Raises:
209 RuntimeError: If connector is not initialized
210 RuntimeError: If change detector is not initialized
211 """
212 if not self._initialized:
213 raise RuntimeError(
214 "Connector not initialized. Use the connector as an async context manager."
215 )
217 try:
218 # Get all pages
219 pages = await self._get_all_pages()
220 self.logger.debug(f"Found {len(pages)} pages to process", pages=pages)
221 yielded_count = 0
223 for page in pages:
224 try:
225 if not self._should_process_url(page):
226 self.logger.debug("Skipping URL", url=page)
227 continue
229 self.logger.debug("Processing URL", url=page)
231 content, title = await self._process_page(page)
232 if (
233 content and content.strip()
234 ): # Only yield documents with non-empty content
235 doc = self._build_page_document(page, content, title)
236 doc_id = doc.id
237 self.logger.debug(
238 "Created document",
239 url=page,
240 content_length=len(content),
241 title=title,
242 doc_id=doc_id,
243 )
244 yield doc
245 yielded_count += 1
246 self.logger.debug(
247 "Document created",
248 url=page,
249 content_length=len(content),
250 title=title,
251 doc_id=doc_id,
252 )
254 # Process attachments if enabled
255 if (
256 self.config.download_attachments
257 and self.attachment_downloader
258 ):
259 # We need to get the HTML again to extract attachments
260 try:
261 try:
262 response = await _aiohttp_request(
263 self.client,
264 "GET",
265 page,
266 rate_limiter=self._rate_limiter,
267 retries=3,
268 backoff_factor=0.5,
269 overall_timeout=60.0,
270 )
271 # Ensure HTTP errors are surfaced consistently
272 response.raise_for_status()
273 except aiohttp.ClientError as e:
274 raise HTTPRequestError(
275 url=page, message=str(e)
276 ) from e
278 html = await _read_text(response)
279 attachment_metadata = self._extract_attachments(
280 html, page, doc_id
281 )
283 if attachment_metadata:
284 self.logger.info(
285 "Processing attachments for PublicDocs page",
286 page_url=page,
287 attachment_count=len(attachment_metadata),
288 )
290 attachment_documents = await self.attachment_downloader.download_and_process_attachments(
291 attachment_metadata, doc
292 )
293 for attachment_doc in attachment_documents:
294 yield attachment_doc
295 yielded_count += 1
297 self.logger.debug(
298 "Processed attachments for PublicDocs page",
299 page_url=page,
300 processed_count=len(attachment_documents),
301 )
302 except Exception as e:
303 self.logger.error(
304 f"Failed to process attachments for page {page}: {e}"
305 )
306 # Continue processing even if attachment processing fails
307 else:
308 self.logger.warning(
309 "Skipping page with empty content",
310 url=page,
311 title=title,
312 )
313 except Exception as e:
314 self.logger.error(f"Failed to process page {page}: {e}")
315 continue
317 if yielded_count == 0:
318 self.logger.warning("No valid documents found to process")
320 except Exception as e:
321 self.logger.error("Failed to get documentation", error=str(e))
322 raise
324 async def get_documents(self) -> list[Document]:
325 """Get documents from PublicDocs (DEPRECATED - use stream_documents)."""
326 return await super().get_documents()
328 async def fetch_by_id(self, entity_id: str) -> Document | None:
329 """Fetch a single documentation page by its URL."""
330 parsed = urlparse(entity_id)
331 base = urlparse(self.base_url)
332 base_path = base.path.rstrip("/")
333 entity_path = parsed.path.rstrip("/")
334 in_scope = (
335 parsed.scheme in {"http", "https"}
336 and parsed.netloc == base.netloc
337 and (entity_path == base_path or entity_path.startswith(base_path + "/"))
338 )
339 if not in_scope or not self._should_process_url(entity_id):
340 return None
341 try:
342 content, title = await self._process_page(entity_id)
343 except Exception as e:
344 self.logger.error(f"Failed to fetch page {entity_id}: {e}")
345 return None
346 if not content or not content.strip():
347 return None
348 return self._build_page_document(entity_id, content, title)
350 async def list_entity_ids(self) -> AsyncIterator[str]:
351 """Stream the URLs of all processable documentation pages."""
352 for page in await self._get_all_pages():
353 if self._should_process_url(page):
354 yield page
356 async def _process_page(self, url: str) -> tuple[str | None, str | None]:
357 """Process a single documentation page.
359 Returns:
360 tuple[str | None, str | None]: A tuple containing (content, title)
362 Raises:
363 ConnectorNotInitializedError: If connector is not initialized
364 HTTPRequestError: If HTTP request fails
365 PageProcessingError: If page processing fails
366 """
367 self.logger.debug("Starting page processing", url=url)
368 try:
369 if not self._initialized:
370 raise ConnectorNotInitializedError(
371 "Connector not initialized. Use async context manager."
372 )
374 self.logger.debug("Making HTTP request", url=url)
375 try:
376 response = await _aiohttp_request(
377 self.client,
378 "GET",
379 url,
380 rate_limiter=self._rate_limiter,
381 retries=3,
382 backoff_factor=0.5,
383 overall_timeout=60.0,
384 )
385 response.raise_for_status() # This is a synchronous method, no need to await
386 except aiohttp.ClientError as e:
387 raise HTTPRequestError(url=url, message=str(e)) from e
389 self.logger.debug(
390 "HTTP request successful", url=url, status_code=response.status
391 )
393 try:
394 # Extract links for crawling
395 self.logger.debug("Extracting links from page", url=url)
396 html = await response.text()
397 links = self._extract_links(html, url)
398 self.logger.info(
399 "Adding new links to queue", url=url, new_links=len(links)
400 )
401 for link in links:
402 if link not in self.visited_urls:
403 self.url_queue.append(link)
405 # Extract title from raw HTML
406 title = self._extract_title(html)
407 self.logger.debug("Extracted title", url=url, title=title)
409 if self.config.content_type == "html":
410 self.logger.debug("Processing Page", url=url)
411 content = self._extract_content(html)
412 self.logger.debug(
413 "HTML content processed",
414 url=url,
415 content_length=len(content) if content else 0,
416 )
417 return content, title
418 else:
419 self.logger.debug("Processing raw content", url=url)
420 self.logger.debug(
421 "Raw content length",
422 url=url,
423 content_length=len(html) if html else 0,
424 )
425 return html, title
426 except Exception as e:
427 raise DocumentProcessingError(
428 f"Failed to process page {url}: {e!s}"
429 ) from e
431 except (
432 ConnectorNotInitializedError,
433 HTTPRequestError,
434 DocumentProcessingError,
435 ):
436 raise
437 except Exception as e:
438 raise ConnectorError(
439 f"Unexpected error processing page {url}: {e!s}"
440 ) from e
442 def _extract_links(self, html: str, current_url: str) -> list[str]:
443 """Extract all links from the HTML content."""
444 self.logger.debug(
445 "Starting link extraction", current_url=current_url, html_length=len(html)
446 )
447 soup = BeautifulSoup(html, "html.parser")
448 links = []
450 for link in soup.find_all("a", href=True):
451 href = str(cast(BeautifulSoup, link)["href"]) # type: ignore
452 # Convert relative URLs to absolute
453 absolute_url = urljoin(current_url, href)
455 # Only include links that are under the base URL
456 if absolute_url.startswith(self.base_url):
457 # Remove fragment identifiers
458 absolute_url = absolute_url.split("#")[0]
459 links.append(absolute_url)
460 self.logger.debug(
461 "Found valid link", original_href=href, absolute_url=absolute_url
462 )
464 self.logger.debug("Link extraction completed", total_links=len(links))
465 return links
467 def _extract_content(self, html: str) -> str:
468 """Extract the main content from HTML using configured selectors."""
469 self.logger.debug("Starting content extraction", html_length=len(html))
470 self.logger.debug("HTML content preview", preview=html[:1000])
471 soup = BeautifulSoup(html, "html.parser")
472 self.logger.debug("HTML parsed successfully")
474 # Log the selectors being used
475 self.logger.debug(
476 "Using selectors",
477 content_selector=self.config.selectors.content,
478 remove_selectors=self.config.selectors.remove,
479 code_blocks_selector=self.config.selectors.code_blocks,
480 )
482 # Remove unwanted elements
483 for selector in self.config.selectors.remove:
484 self.logger.debug(f"Processing selector: {selector}")
485 elements = soup.select(selector)
486 self.logger.debug(
487 f"Found {len(elements)} elements for selector: {selector}"
488 )
489 for element in elements:
490 element.decompose()
492 # Find main content
493 self.logger.debug(
494 f"Looking for main content with selector: {self.config.selectors.content}"
495 )
496 content = soup.select_one(self.config.selectors.content)
497 if not content:
498 self.logger.warning(
499 "Could not find main content using selector",
500 selector=self.config.selectors.content,
501 )
502 # Log the first 1000 characters of the HTML to help debug
503 self.logger.debug("HTML content preview", preview=html[:1000])
504 return ""
506 self.logger.debug(
507 "Found main content element", content_length=len(content.text)
508 )
510 # Preserve code blocks
511 self.logger.debug(
512 f"Looking for code blocks with selector: {self.config.selectors.code_blocks}"
513 )
514 code_blocks = content.select(self.config.selectors.code_blocks)
515 self.logger.debug(f"Found {len(code_blocks)} code blocks")
517 for code_block in code_blocks:
518 code_text = code_block.text
519 if code_text: # Only process non-empty code blocks
520 new_code = BeautifulSoup(f"\n```\n{code_text}\n```\n", "html.parser")
521 if new_code.string: # Ensure we have a valid string to replace with
522 code_block.replace_with(new_code.string) # type: ignore[arg-type]
524 extracted_text = content.get_text(separator="\n", strip=True)
525 self.logger.debug(
526 "Content extraction completed",
527 extracted_length=len(extracted_text),
528 preview=extracted_text[:200] if extracted_text else "",
529 )
530 return extracted_text
532 def _extract_title(self, html: str) -> str:
533 """Extract the title from HTML content."""
534 self.logger.debug("Starting title extraction", html_length=len(html))
535 soup = BeautifulSoup(html, "html.parser")
537 # Production logging: Log title extraction process without verbose HTML content
538 title_tags = soup.find_all("title")
539 if logging.getLogger().isEnabledFor(logging.DEBUG):
540 self.logger.debug(
541 "Found title tags during HTML parsing",
542 count=len(title_tags),
543 html_length=len(html),
544 )
546 # First try to find the title in head/title
547 title_tag = soup.find("title")
548 if title_tag:
549 title = title_tag.get_text(strip=True)
550 self.logger.debug("Found title in title tag", title=title)
551 return title
553 # Then try to find a title in the main content
554 content = soup.select_one(self.config.selectors.content)
555 if content:
556 # Look for h1 in the content
557 h1 = content.find("h1")
558 if h1:
559 title = h1.get_text(strip=True)
560 self.logger.debug("Found title in content", title=title)
561 return title
563 # Look for the first heading
564 heading = content.find(["h1", "h2", "h3", "h4", "h5", "h6"])
565 if heading:
566 title = heading.get_text(strip=True)
567 self.logger.debug("Found title in heading", title=title)
568 return title
570 # If no title found, use a default
571 default_title = "Untitled Document"
572 self.logger.warning(
573 "No title found, using default", default_title=default_title
574 )
575 return default_title
577 def _extract_attachments(
578 self, html: str, page_url: str, document_id: str
579 ) -> list[AttachmentMetadata]:
580 """Extract attachment links from HTML content.
582 Args:
583 html: HTML content to parse
584 page_url: URL of the current page
585 document_id: ID of the parent document
587 Returns:
588 List of attachment metadata objects
589 """
590 if not self.config.download_attachments:
591 return []
593 self.logger.debug("Starting attachment extraction", page_url=page_url)
594 soup = BeautifulSoup(html, "html.parser")
595 attachments = []
597 # Use configured selectors to find attachment links
598 for selector in self.config.attachment_selectors:
599 links = soup.select(selector)
600 self.logger.debug(f"Found {len(links)} links for selector: {selector}")
602 for link in links:
603 href = link.get("href")
604 if not href:
605 continue
607 # Convert relative URLs to absolute
608 absolute_url = urljoin(page_url, str(href))
610 # Extract filename from URL
611 parsed_url = urlparse(absolute_url)
612 filename = (
613 parsed_url.path.split("/")[-1] if parsed_url.path else "unknown"
614 )
616 # Try to determine file extension and MIME type
617 file_ext = filename.split(".")[-1].lower() if "." in filename else ""
618 mime_type = self._get_mime_type_from_extension(file_ext)
620 # Create attachment metadata
621 attachment = AttachmentMetadata(
622 id=f"{document_id}_{len(attachments)}", # Simple ID generation
623 filename=filename,
624 size=0, # We don't know the size until we download
625 mime_type=mime_type,
626 download_url=absolute_url,
627 parent_document_id=document_id,
628 created_at=None,
629 updated_at=None,
630 author=None,
631 )
632 attachments.append(attachment)
634 self.logger.debug(
635 "Found attachment",
636 filename=filename,
637 url=absolute_url,
638 mime_type=mime_type,
639 )
641 self.logger.debug(f"Extracted {len(attachments)} attachments from page")
642 return attachments
644 def _get_mime_type_from_extension(self, extension: str) -> str:
645 """Get MIME type from file extension.
647 Args:
648 extension: File extension (without dot)
650 Returns:
651 MIME type string
652 """
653 mime_types = {
654 "pdf": "application/pdf",
655 "docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
656 "xls": "application/vnd.ms-excel",
657 "xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
658 "pptx": "application/vnd.openxmlformats-officedocument.presentationml.presentation",
659 "txt": "text/plain",
660 "csv": "text/csv",
661 "json": "application/json",
662 "xml": "application/xml",
663 "zip": "application/zip",
664 }
665 return mime_types.get(extension, "application/octet-stream")
667 async def _get_all_pages(self) -> list[str]:
668 """Get all pages from the source.
670 Returns:
671 List of page URLs
673 Raises:
674 ConnectorNotInitializedError: If connector is not initialized
675 HTTPRequestError: If HTTP request fails
676 PublicDocsConnectorError: If page discovery fails
677 """
678 if not self._initialized:
679 raise ConnectorNotInitializedError(
680 "Connector not initialized. Use async context manager."
681 )
683 try:
684 self.logger.debug(
685 "Fetching pages from base URL",
686 base_url=str(self.config.base_url),
687 path_pattern=self.config.path_pattern,
688 )
690 # Reuse existing client if available; otherwise, create a temporary session
691 if getattr(self, "_client", None):
692 client = self.client
693 try:
694 return await _discover_pages(
695 client,
696 str(self.config.base_url),
697 path_pattern=self.config.path_pattern,
698 exclude_paths=self.config.exclude_paths,
699 logger=self.logger,
700 )
701 except aiohttp.ClientError as e:
702 raise HTTPRequestError(
703 url=str(self.config.base_url), message=str(e)
704 ) from e
705 except Exception as e:
706 raise ConnectorError(
707 f"Failed to process page content: {e!s}"
708 ) from e
709 else:
710 async with aiohttp.ClientSession() as client:
711 try:
712 return await _discover_pages(
713 client,
714 str(self.config.base_url),
715 path_pattern=self.config.path_pattern,
716 exclude_paths=self.config.exclude_paths,
717 logger=self.logger,
718 )
719 except aiohttp.ClientError as e:
720 raise HTTPRequestError(
721 url=str(self.config.base_url), message=str(e)
722 ) from e
723 except Exception as e:
724 raise ConnectorError(
725 f"Failed to process page content: {e!s}"
726 ) from e
728 except (ConnectorNotInitializedError, HTTPRequestError, ConnectorError):
729 raise
730 except Exception as e:
731 raise ConnectorError(f"Unexpected error getting pages: {e!s}") from e