Coverage for src/qdrant_loader/connectors/confluence/connector.py: 71%
399 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-20 10:15 +0000
1import re
2from collections.abc import AsyncIterator
3from datetime import datetime
4from urllib.parse import quote, urljoin
6import requests
8from qdrant_loader.config.types import SourceType
9from qdrant_loader.connectors.base import BaseConnector
10from qdrant_loader.connectors.confluence.auth import (
11 auto_detect_deployment_type as _auto_detect_type,
12)
13from qdrant_loader.connectors.confluence.auth import setup_authentication as _setup_auth
14from qdrant_loader.connectors.confluence.config import (
15 ConfluenceDeploymentType,
16 ConfluenceSpaceConfig,
17)
18from qdrant_loader.connectors.confluence.mappers import (
19 extract_hierarchy_info as _extract_hierarchy_info_helper,
20)
21from qdrant_loader.connectors.confluence.pagination import (
22 CONTENT_EXPAND as _CONTENT_EXPAND,
23)
24from qdrant_loader.connectors.confluence.pagination import (
25 build_cloud_search_params as _build_cloud_params,
26)
27from qdrant_loader.connectors.confluence.pagination import (
28 build_dc_search_params as _build_dc_params,
29)
30from qdrant_loader.connectors.shared.attachments import AttachmentReader
31from qdrant_loader.connectors.shared.attachments.metadata import (
32 confluence_attachment_to_metadata,
33)
34from qdrant_loader.connectors.shared.http import (
35 RateLimiter,
36)
37from qdrant_loader.connectors.shared.http import (
38 request_with_policy as _http_request_with_policy,
39)
40from qdrant_loader.core.attachment_downloader import AttachmentMetadata
41from qdrant_loader.core.document import Document
42from qdrant_loader.core.file_conversion import (
43 FileConversionConfig,
44 FileConverter,
45 FileDetector,
46)
47from qdrant_loader.utils.logging import LoggingConfig
49logger = LoggingConfig.get_logger(__name__)
52class ConfluenceConnector(BaseConnector):
53 """Connector for Atlassian Confluence."""
55 def __init__(self, config: ConfluenceSpaceConfig):
56 """Initialize the connector with configuration.
58 Args:
59 config: Confluence configuration
60 """
61 super().__init__(config)
62 self.config = config
63 self.base_url = config.base_url
65 # Initialize session
66 self.session = requests.Session()
67 # Rate limiter (configurable RPM)
68 self._rate_limiter = RateLimiter.per_minute(
69 getattr(self.config, "requests_per_minute", 60)
70 )
72 # Set up authentication based on deployment type
73 self._setup_authentication()
74 self._initialized = False
76 # Initialize file conversion and attachment handling components
77 self.file_converter = None
78 self.file_detector = None
79 self.attachment_downloader = None
81 if self.config.enable_file_conversion:
82 logger.info("File conversion enabled for Confluence connector")
83 # File conversion config will be set from global config during ingestion
84 self.file_detector = FileDetector()
85 else:
86 logger.debug("File conversion disabled for Confluence connector")
88 def set_file_conversion_config(self, file_conversion_config: FileConversionConfig):
89 """Set file conversion configuration from global config.
91 Args:
92 file_conversion_config: Global file conversion configuration
93 """
94 if self.config.enable_file_conversion:
95 self.file_converter = FileConverter(file_conversion_config)
97 # Initialize attachment downloader if download_attachments is enabled
98 if self.config.download_attachments:
99 from qdrant_loader.core.attachment_downloader import (
100 AttachmentDownloader,
101 )
103 downloader = AttachmentDownloader(
104 session=self.session,
105 file_conversion_config=file_conversion_config,
106 enable_file_conversion=True,
107 max_attachment_size=file_conversion_config.max_file_size,
108 )
109 self.attachment_downloader = AttachmentReader(
110 session=self.session, downloader=downloader
111 )
112 logger.info("Attachment reader initialized with file conversion")
113 else:
114 logger.debug("Attachment downloading disabled")
116 logger.debug("File converter initialized with global config")
118 def _setup_authentication(self):
119 """Set up authentication based on deployment type."""
120 _setup_auth(self.session, self.config)
122 def _auto_detect_deployment_type(self) -> ConfluenceDeploymentType:
123 """Auto-detect the Confluence deployment type based on the base URL.
125 Returns:
126 ConfluenceDeploymentType: Detected deployment type
127 """
128 return _auto_detect_type(str(self.base_url))
130 async def __aenter__(self):
131 """Async context manager entry."""
132 if not self._initialized:
133 self._initialized = True
134 return self
136 async def __aexit__(self, exc_type, exc_val, _exc_tb):
137 """Async context manager exit."""
138 self._initialized = False
140 def _get_api_url(self, endpoint: str) -> str:
141 """Construct the full API URL for an endpoint.
143 Args:
144 endpoint: API endpoint path
146 Returns:
147 str: Full API URL
148 """
149 return f"{self.base_url}/rest/api/{endpoint}"
151 async def _make_request(self, method: str, endpoint: str, **kwargs) -> dict:
152 """Make an authenticated request to the Confluence API.
154 Args:
155 method: HTTP method
156 endpoint: API endpoint path
157 **kwargs: Additional request parameters
159 Returns:
160 dict: Response data
162 Raises:
163 requests.exceptions.RequestException: If the request fails
164 """
165 url = self._get_api_url(endpoint)
166 try:
167 if not self.session.headers.get("Authorization"):
168 kwargs["auth"] = self.session.auth
170 response = await _http_request_with_policy(
171 self.session,
172 method,
173 url,
174 rate_limiter=self._rate_limiter,
175 retries=3,
176 backoff_factor=0.5,
177 status_forcelist=(429, 500, 502, 503, 504),
178 overall_timeout=90.0,
179 **kwargs,
180 )
181 response.raise_for_status()
182 return response.json()
183 except requests.exceptions.RequestException as e:
184 logger.error(f"Failed to make request to {url}: {e}")
185 logger.error(
186 "Request details",
187 method=method,
188 url=url,
189 deployment_type=self.config.deployment_type,
190 has_auth_header=bool(self.session.headers.get("Authorization")),
191 has_session_auth=bool(self.session.auth),
192 )
193 raise
195 async def _get_space_content_cloud(
196 self, cursor: str | None = None, light: bool = False
197 ) -> dict:
198 """Fetch content from a Confluence Cloud space using cursor-based pagination.
200 Args:
201 cursor: Cursor for pagination. If None, starts from the beginning.
202 light: If True, request a minimal `expand` (no page bodies) for
203 cheap ID-only listing.
205 Returns:
206 dict: Response containing space content
207 """
208 # Build params via helper
209 params = _build_cloud_params(
210 self.config.space_key, self.config.content_types, cursor, light=light
211 )
213 logger.debug(
214 "Making Confluence Cloud API request",
215 url=f"{self.base_url}/rest/api/content/search",
216 params=params,
217 )
218 response = await self._make_request("GET", "content/search", params=params)
219 if response and "results" in response:
220 # For Cloud, we can't easily calculate page numbers from cursor, so just log occasionally
221 if len(response["results"]) > 0:
222 logger.debug(
223 f"Fetching Confluence Cloud documents: {len(response['results'])} found",
224 count=len(response["results"]),
225 total_size=response.get("totalSize", response.get("size", 0)),
226 )
227 return response
229 async def _get_space_content_datacenter(
230 self, start: int = 0, light: bool = False
231 ) -> dict:
232 """Fetch content from a Confluence Data Center space using start/limit pagination.
234 Args:
235 start: Starting index for pagination. Defaults to 0.
236 light: If True, request a minimal `expand` (no page bodies) for
237 cheap ID-only listing.
239 Returns:
240 dict: Response containing space content
241 """
242 params = _build_dc_params(
243 self.config.space_key, self.config.content_types, start, light=light
244 )
246 logger.debug(
247 "Making Confluence Data Center API request",
248 url=f"{self.base_url}/rest/api/content/search",
249 params=params,
250 )
251 response = await self._make_request("GET", "content/search", params=params)
252 if response and "results" in response:
253 # Only log every 10th page to reduce verbosity
254 page_num = start // 25 + 1
255 if page_num == 1 or page_num % 10 == 0:
256 logger.debug(
257 f"Fetching Confluence Data Center documents (page {page_num}): {len(response['results'])} found",
258 count=len(response["results"]),
259 total_size=response.get("totalSize", response.get("size", 0)),
260 start=start,
261 )
262 return response
264 async def _get_space_content(self, start: int = 0) -> dict:
265 """Backward compatibility method for tests.
267 Args:
268 start: Starting index for pagination. Defaults to 0.
270 Returns:
271 dict: Response containing space content
272 """
273 if self.config.deployment_type == ConfluenceDeploymentType.CLOUD:
274 # For Cloud, ignore start parameter and use cursor=None
275 return await self._get_space_content_cloud(None)
276 else:
277 # For Data Center, use start parameter
278 return await self._get_space_content_datacenter(start)
280 async def _get_content_attachments(
281 self, content_id: str
282 ) -> list[AttachmentMetadata]:
283 """Fetch attachments for a specific content item.
285 Args:
286 content_id: ID of the content item
288 Returns:
289 List of attachment metadata
290 """
291 if not self.config.download_attachments:
292 return []
294 try:
295 # Fetch attachments using Confluence API
296 endpoint = f"content/{content_id}/child/attachment"
297 params = {
298 "expand": "metadata,version,history", # Include history for better metadata
299 "limit": 50, # Reasonable limit for attachments per page
300 }
302 response = await self._make_request("GET", endpoint, params=params)
303 attachments = []
305 for attachment_data in response.get("results", []):
306 try:
307 translated = confluence_attachment_to_metadata(
308 attachment_data,
309 base_url=str(self.base_url),
310 parent_id=content_id,
311 )
312 if translated is None:
313 logger.warning(
314 "No download link found for attachment",
315 attachment_id=attachment_data.get("id"),
316 filename=attachment_data.get("title"),
317 deployment_type=self.config.deployment_type,
318 )
319 continue
321 attachments.append(translated)
323 logger.debug(
324 "Found attachment",
325 attachment_id=getattr(translated, "id", None),
326 filename=getattr(translated, "filename", None),
327 size=getattr(translated, "size", None),
328 mime_type=getattr(translated, "mime_type", None),
329 deployment_type=self.config.deployment_type,
330 )
332 except Exception as e:
333 logger.warning(
334 "Failed to process attachment metadata",
335 attachment_id=attachment_data.get("id"),
336 filename=attachment_data.get("title"),
337 deployment_type=self.config.deployment_type,
338 error=str(e),
339 )
340 continue
342 logger.debug(
343 "Found attachments for content",
344 content_id=content_id,
345 attachment_count=len(attachments),
346 deployment_type=self.config.deployment_type,
347 )
349 return attachments
351 except Exception as e:
352 logger.error(
353 "Failed to fetch attachments",
354 content_id=content_id,
355 deployment_type=self.config.deployment_type,
356 error=str(e),
357 )
358 return []
360 async def _process_attachments_for_document(
361 self, content: dict, document: Document
362 ) -> list[Document]:
363 """Process attachments for a given content item and parent document.
365 Checks configuration flags and uses the attachment downloader to
366 fetch and convert attachments into child documents.
368 Args:
369 content: Confluence content item
370 document: Parent document corresponding to the content item
372 Returns:
373 List of generated attachment documents (may be empty)
374 """
375 if not (self.config.download_attachments and self.attachment_downloader):
376 return []
378 try:
379 content_id = content.get("id")
380 attachments = await self._get_content_attachments(content_id)
381 if not attachments:
382 return []
384 attachment_docs = await self.attachment_downloader.fetch_and_process(
385 attachments, document
386 )
387 logger.debug(
388 f"Processed {len(attachment_docs)} attachments for {content.get('type')} '{content.get('title')}'",
389 content_id=content.get("id"),
390 )
391 return attachment_docs
392 except Exception as e:
393 logger.error(
394 f"Failed to process attachments for {content.get('type')} '{content.get('title')}' (ID: {content.get('id')}): {e!s}"
395 )
396 return []
398 def _should_process_content(self, content: dict) -> bool:
399 """Check if content should be processed based on labels.
401 Args:
402 content: Content metadata from Confluence API
404 Returns:
405 bool: True if content should be processed, False otherwise
406 """
407 # Get content labels
408 labels = {
409 label["name"]
410 for label in content.get("metadata", {})
411 .get("labels", {})
412 .get("results", [])
413 }
415 # Log content details for debugging
416 logger.debug(
417 "Checking content for processing",
418 content_id=content.get("id"),
419 content_type=content.get("type"),
420 title=content.get("title"),
421 labels=labels,
422 exclude_labels=self.config.exclude_labels,
423 include_labels=self.config.include_labels,
424 )
426 # Check exclude labels first, if there are any specified
427 if self.config.exclude_labels and any(
428 label in labels for label in self.config.exclude_labels
429 ):
430 logger.debug(
431 "Content excluded due to exclude labels",
432 content_id=content.get("id"),
433 title=content.get("title"),
434 matching_labels=[
435 label for label in labels if label in self.config.exclude_labels
436 ],
437 )
438 return False
440 # If include labels are specified, content must have at least one
441 if self.config.include_labels:
442 has_include_label = any(
443 label in labels for label in self.config.include_labels
444 )
445 if not has_include_label:
446 logger.debug(
447 "Content excluded due to missing include labels",
448 content_id=content.get("id"),
449 title=content.get("title"),
450 required_labels=self.config.include_labels,
451 )
452 return has_include_label
454 return True
456 def _extract_hierarchy_info(self, content: dict) -> dict:
457 """Extract page hierarchy information from Confluence content.
459 Args:
460 content: Content item from Confluence API
462 Returns:
463 dict: Hierarchy information including ancestors, parent, and children
464 """
465 return _extract_hierarchy_info_helper(content)
467 def _process_content(
468 self, content: dict, clean_html: bool = True
469 ) -> Document | None:
470 """Process a single content item from Confluence.
472 Args:
473 content: Content item from Confluence API
474 clean_html: Whether to clean HTML tags from content. Defaults to True.
476 Returns:
477 Document if processing successful
479 Raises:
480 ValueError: If required fields are missing or malformed
481 """
482 try:
483 # Extract required fields
484 content_id = content.get("id")
485 title = content.get("title")
486 space = content.get("space", {}).get("key")
488 # Log content details for debugging
489 logger.debug(
490 "Processing content",
491 content_id=content_id,
492 title=title,
493 space=space,
494 type=content.get("type"),
495 version=content.get("version", {}).get("number"),
496 has_body=bool(content.get("body", {}).get("storage", {}).get("value")),
497 comment_count=len(
498 content.get("children", {}).get("comment", {}).get("results", [])
499 ),
500 label_count=len(
501 content.get("metadata", {}).get("labels", {}).get("results", [])
502 ),
503 )
505 body = content.get("body", {}).get("storage", {}).get("value")
506 # Check for missing or malformed body
507 if not body:
508 logger.warning(
509 "Content body is missing or malformed, using title as content",
510 content_id=content_id,
511 title=title,
512 content_type=content.get("type"),
513 space=space,
514 )
515 # Use title as fallback content instead of failing
516 body = title or f"[Empty page: {content_id}]"
518 # Check for other missing required fields
519 missing_fields = []
520 if not content_id:
521 missing_fields.append("id")
522 if not title:
523 missing_fields.append("title")
524 if not space:
525 missing_fields.append("space")
527 if missing_fields:
528 logger.warning(
529 "Content is missing required fields",
530 content_id=content_id,
531 title=title,
532 content_type=content.get("type"),
533 missing_fields=missing_fields,
534 space=space,
535 )
536 raise ValueError(
537 f"Content is missing required fields: {', '.join(missing_fields)}"
538 )
540 # Get version information
541 version = content.get("version", {})
542 version_number = (
543 version.get("number", 1) if isinstance(version, dict) else 1
544 )
546 # Get author information with better error handling
547 author = None
548 try:
549 author = (
550 content.get("history", {}).get("createdBy", {}).get("displayName")
551 )
552 if not author:
553 # Fallback to version author for Data Center
554 author = content.get("version", {}).get("by", {}).get("displayName")
555 except (AttributeError, TypeError):
556 logger.debug(
557 "Could not extract author information", content_id=content_id
558 )
560 # Get timestamps with improved parsing for both Cloud and Data Center
561 created_at = None
562 updated_at = None
564 # Try to get creation date from history (both Cloud and Data Center)
565 try:
566 if "history" in content and "createdDate" in content["history"]:
567 created_at = content["history"]["createdDate"]
568 elif "history" in content and "createdAt" in content["history"]:
569 # Alternative field name in some Data Center versions
570 created_at = content["history"]["createdAt"]
571 except (ValueError, TypeError, KeyError):
572 logger.debug("Could not parse creation date", content_id=content_id)
574 # Try to get update date from version (both Cloud and Data Center)
575 try:
576 if "version" in content and "when" in content["version"]:
577 updated_at = content["version"]["when"]
578 elif "version" in content and "friendlyWhen" in content["version"]:
579 # Some Data Center versions use friendlyWhen
580 updated_at = content["version"]["friendlyWhen"]
581 except (ValueError, TypeError, KeyError):
582 logger.debug("Could not parse update date", content_id=content_id)
584 # Process comments
585 comments = []
586 if "children" in content and "comment" in content["children"]:
587 for comment in content["children"]["comment"]["results"]:
588 comment_body = (
589 comment.get("body", {}).get("storage", {}).get("value", "")
590 )
591 comment_author = (
592 comment.get("history", {})
593 .get("createdBy", {})
594 .get("displayName", "")
595 )
596 comment_created = comment.get("history", {}).get("createdDate", "")
597 comments.append(
598 {
599 "body": (
600 self._clean_html(comment_body)
601 if clean_html
602 else comment_body
603 ),
604 "author": comment_author,
605 "created_at": comment_created,
606 }
607 )
609 # Extract hierarchy information
610 hierarchy_info = self._extract_hierarchy_info(content)
612 # Build canonical and display URLs
613 canonical_url = self._construct_canonical_page_url(
614 space or "",
615 content_id or "",
616 content.get("type", "page"),
617 )
618 display_url = self._construct_page_url(
619 space or "",
620 content_id or "",
621 title or "",
622 content.get("type", "page"),
623 )
625 # Create metadata with all available information including hierarchy
626 metadata = {
627 "id": content_id,
628 "title": title,
629 "space": space,
630 "version": version_number,
631 "type": content.get("type", "unknown"),
632 "author": author,
633 # Human-friendly URL (kept in metadata)
634 "display_url": display_url,
635 "labels": [
636 label["name"]
637 for label in content.get("metadata", {})
638 .get("labels", {})
639 .get("results", [])
640 ],
641 "comments": comments,
642 "updated_at": updated_at,
643 "created_at": created_at,
644 # Page hierarchy information
645 "hierarchy": hierarchy_info,
646 "parent_id": hierarchy_info["parent_id"],
647 "parent_title": hierarchy_info["parent_title"],
648 "ancestors": hierarchy_info["ancestors"],
649 "children": hierarchy_info["children"],
650 "depth": hierarchy_info["depth"],
651 "breadcrumb": hierarchy_info["breadcrumb"],
652 "breadcrumb_text": (
653 " > ".join(hierarchy_info["breadcrumb"])
654 if hierarchy_info["breadcrumb"]
655 else ""
656 ),
657 }
659 # Clean content if requested
660 content_text = self._clean_html(body) if clean_html else body
662 # Parse timestamps for Document constructor
663 parsed_created_at = self._parse_timestamp(created_at)
664 parsed_updated_at = self._parse_timestamp(updated_at)
666 # Create document with all fields properly populated
667 document = Document(
668 title=title,
669 content=content_text,
670 content_type="html",
671 metadata=metadata,
672 source_type=SourceType.CONFLUENCE,
673 source=self.config.source,
674 url=canonical_url,
675 is_deleted=False,
676 updated_at=parsed_updated_at,
677 created_at=parsed_created_at,
678 )
680 return document
682 except Exception as e:
683 logger.error(
684 "Failed to process content",
685 content_id=content.get("id"),
686 content_title=content.get("title"),
687 content_type=content.get("type"),
688 error=str(e),
689 error_type=type(e).__name__,
690 )
691 raise
693 def _construct_page_url(
694 self, space: str, content_id: str, title: str, content_type: str = "page"
695 ) -> str:
696 """Construct the appropriate URL for a Confluence page based on deployment type.
698 Args:
699 space: The space key
700 content_id: The content ID
701 title: The page title (used for Data Center URLs)
702 content_type: The type of content (page, blogpost, etc.)
704 Returns:
705 The constructed URL
706 """
707 # Ensure base is treated as a directory to preserve any path components
708 base = str(self.base_url)
709 base = base if base.endswith("/") else base + "/"
711 if self.config.deployment_type == ConfluenceDeploymentType.CLOUD:
712 # Cloud URLs use ID-based format
713 if content_type == "blogpost":
714 path = f"spaces/{space}/blog/{content_id}"
715 else:
716 path = f"spaces/{space}/pages/{content_id}"
717 return urljoin(base, path)
718 else:
719 # Data Center/Server URLs - use title for better readability
720 # URL-encode the title, replacing spaces with '+' (Confluence format)
721 encoded_title = quote(title.replace(" ", "+"), safe="+")
722 if content_type == "blogpost":
723 path = f"display/{space}/{encoded_title}"
724 else:
725 path = f"display/{space}/{encoded_title}"
726 return urljoin(base, path)
728 def _construct_canonical_page_url(
729 self, space: str, content_id: str, content_type: str = "page"
730 ) -> str:
731 """Construct a canonical ID-based URL for both Cloud and Data Center."""
732 base = str(self.base_url)
733 base = base if base.endswith("/") else base + "/"
735 if content_type == "blogpost":
736 path = f"spaces/{space}/blog/{content_id}"
737 else:
738 path = f"spaces/{space}/pages/{content_id}"
739 return urljoin(base, path)
741 def _parse_timestamp(self, timestamp_str: str | None) -> "datetime | None":
742 """Parse a timestamp string into a datetime object.
744 Args:
745 timestamp_str: The timestamp string to parse
747 Returns:
748 Parsed datetime object or None if parsing fails
749 """
750 if not timestamp_str:
751 return None
753 try:
754 import re
755 from datetime import datetime
757 # Handle various timestamp formats from Confluence
758 # ISO format with timezone: 2024-05-24T20:57:56.130+07:00
759 if re.match(
760 r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}[+-]\d{2}:\d{2}",
761 timestamp_str,
762 ):
763 return datetime.fromisoformat(timestamp_str)
765 # ISO format without microseconds: 2024-05-24T20:57:56+07:00
766 elif re.match(
767 r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}[+-]\d{2}:\d{2}", timestamp_str
768 ):
769 return datetime.fromisoformat(timestamp_str)
771 # ISO format with Z timezone: 2024-05-24T20:57:56.130Z
772 elif re.match(
773 r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z", timestamp_str
774 ):
775 return datetime.fromisoformat(timestamp_str.replace("Z", "+00:00"))
777 # ISO format without timezone: 2024-05-24T20:57:56.130
778 elif re.match(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}", timestamp_str):
779 return datetime.fromisoformat(timestamp_str)
781 # Fallback: try direct parsing
782 else:
783 return datetime.fromisoformat(timestamp_str)
785 except (ValueError, TypeError, AttributeError) as e:
786 logger.debug(f"Failed to parse timestamp '{timestamp_str}': {e}")
787 return None
789 def _clean_html(self, html: str) -> str:
790 """Clean HTML content by removing tags and special characters.
792 Args:
793 html: HTML content to clean
795 Returns:
796 Cleaned text
797 """
798 # Remove HTML tags
799 text = re.sub(r"<[^>]+>", " ", html)
800 # Replace HTML entities
801 text = text.replace("&", "and")
802 text = re.sub(r"&[^;]+;", " ", text)
803 # Replace multiple spaces with single space
804 text = re.sub(r"\s+", " ", text)
805 return text.strip()
807 async def _stream_content_cloud(self) -> AsyncIterator[Document]:
808 """Stream documents from Confluence Cloud using cursor-based pagination."""
809 cursor = None
810 page_count = 0
812 while True:
813 try:
814 page_count += 1
815 logger.debug(
816 f"Fetching page {page_count} of Confluence content (cursor={cursor})"
817 )
818 response = await self._get_space_content_cloud(cursor)
819 results = response.get("results", [])
821 if not results:
822 logger.debug("No more results found, ending pagination")
823 break
825 logger.debug(
826 f"Processing {len(results)} documents from page {page_count}"
827 )
829 # Process each content item
830 for content in results:
831 if self._should_process_content(content):
832 try:
833 document = self._process_content(content, clean_html=True)
834 if document:
835 yield document
837 attachment_docs = (
838 await self._process_attachments_for_document(
839 content, document
840 )
841 )
842 for attachment_doc in attachment_docs:
843 yield attachment_doc
845 logger.debug(
846 f"Processed {content['type']} '{content['title']}' "
847 f"(ID: {content['id']}) from space {self.config.space_key}"
848 )
849 except Exception as e:
850 logger.error(
851 "Failed to process Confluence content",
852 content_type=content.get("type"),
853 content_title=content.get("title"),
854 content_id=content.get("id"),
855 error=str(e),
856 )
858 # Get the next cursor from the response
859 next_url = response.get("_links", {}).get("next")
860 if not next_url:
861 logger.debug("No next page link found, ending pagination")
862 break
864 # Extract just the cursor value from the URL
865 try:
866 from urllib.parse import parse_qs, urlparse
868 parsed_url = urlparse(next_url)
869 query_params = parse_qs(parsed_url.query)
870 cursor = query_params.get("cursor", [None])[0]
871 if not cursor:
872 logger.debug("No cursor found in next URL, ending pagination")
873 break
874 logger.debug(f"Found next cursor: {cursor}")
875 except (ValueError, KeyError, AttributeError) as e:
876 logger.error(f"Failed to parse next URL: {e!s}")
877 break
879 except (requests.exceptions.RequestException, ValueError) as e:
880 logger.error(
881 f"Failed to fetch content from space {self.config.space_key}: {e!s}"
882 )
883 raise
885 async def _stream_content_datacenter(self) -> AsyncIterator[Document]:
886 """Stream documents from Confluence Data Center/Server using start/limit pagination."""
887 start = 0
888 limit = 25
889 page_count = 0
891 while True:
892 try:
893 page_count += 1
894 logger.debug(
895 f"Fetching page {page_count} of Confluence content (start={start})"
896 )
897 response = await self._get_space_content_datacenter(start)
898 results = response.get("results", [])
900 if not results:
901 logger.debug("No more results found, ending pagination")
902 break
904 logger.debug(
905 f"Processing {len(results)} documents from page {page_count}"
906 )
908 # Process each content item
909 for content in results:
910 if self._should_process_content(content):
911 try:
912 document = self._process_content(content, clean_html=True)
913 if document:
914 yield document
916 attachment_docs = (
917 await self._process_attachments_for_document(
918 content, document
919 )
920 )
921 for attachment_doc in attachment_docs:
922 yield attachment_doc
924 logger.debug(
925 f"Processed {content['type']} '{content['title']}' "
926 f"(ID: {content['id']}) from space {self.config.space_key}"
927 )
928 except Exception as e:
929 logger.error(
930 "Failed to process Confluence content",
931 content_type=content.get("type"),
932 content_title=content.get("title"),
933 content_id=content.get("id"),
934 error=str(e),
935 )
937 # Check if there are more pages
938 total_size = response.get("totalSize", response.get("size", 0))
939 if start + limit >= total_size:
940 logger.debug(
941 f"Reached end of results: {start + limit} >= {total_size}"
942 )
943 break
945 # Move to next page
946 start += limit
947 logger.debug(f"Moving to next page with start={start}")
949 except (requests.exceptions.RequestException, ValueError) as e:
950 logger.error(
951 f"Failed to fetch content from space {self.config.space_key}: {e!s}"
952 )
953 raise
955 async def stream_documents(
956 self, since: datetime | None = None
957 ) -> AsyncIterator[Document]:
958 """Stream documents from Confluence (WS-1 connector contract)."""
959 count = 0
961 if self.config.deployment_type == ConfluenceDeploymentType.CLOUD:
962 async for document in self._stream_content_cloud():
963 count += 1
964 yield document
965 else:
966 async for document in self._stream_content_datacenter():
967 count += 1
968 yield document
970 logger.info(
971 f"📄 Confluence: {count} documents from space {self.config.space_key}"
972 )
974 async def get_documents(self) -> list[Document]:
975 """Get documents from Confluence (DEPRECATED - use stream_documents)."""
976 return await super().get_documents()
978 async def fetch_by_id(self, entity_id: str) -> Document | None:
979 """Fetch a single Confluence content item by its content ID."""
980 try:
981 params = {"expand": _CONTENT_EXPAND}
982 safe_entity_id = quote(entity_id, safe="")
983 content = await self._make_request(
984 "GET", f"content/{safe_entity_id}", params=params
985 )
986 if not content or not self._should_process_content(content):
987 return None
988 return self._process_content(content, clean_html=True)
989 except Exception as e:
990 logger.error(
991 "Failed to fetch Confluence content by id",
992 content_id=entity_id,
993 error=str(e),
994 )
995 return None
997 async def _stream_content_ids_cloud(self) -> AsyncIterator[str]:
998 """Stream content IDs from Confluence Cloud using a lightweight expand."""
999 cursor = None
1001 while True:
1002 response = await self._get_space_content_cloud(cursor, light=True)
1003 results = response.get("results", [])
1004 if not results:
1005 break
1007 for content in results:
1008 if self._should_process_content(content):
1009 yield content["id"]
1011 next_url = response.get("_links", {}).get("next")
1012 if not next_url:
1013 break
1015 try:
1016 from urllib.parse import parse_qs, urlparse
1018 parsed_url = urlparse(next_url)
1019 query_params = parse_qs(parsed_url.query)
1020 cursor = query_params.get("cursor", [None])[0]
1021 if not cursor:
1022 break
1023 except (ValueError, KeyError, AttributeError):
1024 break
1026 async def _stream_content_ids_datacenter(self) -> AsyncIterator[str]:
1027 """Stream content IDs from Confluence Data Center using a lightweight expand."""
1028 start = 0
1029 limit = 25
1031 while True:
1032 response = await self._get_space_content_datacenter(start, light=True)
1033 results = response.get("results", [])
1034 if not results:
1035 break
1037 for content in results:
1038 if self._should_process_content(content):
1039 yield content["id"]
1041 total_size = response.get("totalSize", response.get("size", 0))
1042 if start + limit >= total_size:
1043 break
1044 start += limit
1046 async def list_entity_ids(self) -> AsyncIterator[str]:
1047 """Stream content IDs for all processable content in the configured space."""
1048 if self.config.deployment_type == ConfluenceDeploymentType.CLOUD:
1049 async for content_id in self._stream_content_ids_cloud():
1050 yield content_id
1051 else:
1052 async for content_id in self._stream_content_ids_datacenter():
1053 yield content_id