Confluence Datasource

class ConfluenceClientFactory(SingletonFactory):
    """
    Factory for creating and managing Confluence client instances.

    This factory ensures only one Confluence client is created per configuration,
    following the singleton pattern provided by the parent SingletonFactory class.
    """

    _configuration_class: Type = ConfluenceDatasourceConfiguration

    @classmethod
    def _create_instance(
        cls, configuration: ConfluenceDatasourceConfiguration
    ) -> Confluence:
        """
        Creates a new Confluence client instance using the provided configuration.

        Args:
            configuration: Configuration object containing Confluence connection details
                          including base URL, username, and password.

        Returns:
            A configured Confluence client instance ready for API interactions.
        """
        return Confluence(
            url=configuration.base_url,
            username=configuration.secrets.username.get_secret_value(),
            password=configuration.secrets.password.get_secret_value(),
        )

class ConfluenceDatasourceConfiguration(DatasourceConfiguration):
    class Secrets(BaseSecrets):
        model_config = ConfigDict(
            env_file_encoding="utf-8",
            env_prefix="RAG__DATASOURCES__CONFLUENCE__",
            env_nested_delimiter="__",
            extra="ignore",
        )

        username: SecretStr = Field(
            ...,
            description="Username credential used to authenticate with the Confluence instance",
        )
        password: SecretStr = Field(
            ...,
            description="Password credential used to authenticate with the Confluence instance",
        )

    host: str = Field(
        "127.0.0.1",
        description="Hostname or IP address of the Confluence server instance",
    )
    protocol: Union[Literal["http"], Literal["https"]] = Field(
        "http",
        description="Communication protocol used to connect to the Confluence server",
    )
    name: Literal[DatasourceName.CONFLUENCE] = Field(
        ...,
        description="Identifier specifying this configuration is for a Confluence datasource",
    )
    secrets: Secrets = Field(
        None,
        description="Authentication credentials required to access the Confluence instance",
    )

    @property
    def base_url(self) -> str:
        """
        Constructs the complete base URL for the Confluence API from the protocol and host.

        Returns:
            str: The fully formed base URL to the Confluence instance
        """
        return f"{self.protocol}://{self.host}"

class ConfluenceDocument(BaseDocument):
    """Document representation for Confluence page content.

    Extends BaseDocument to handle Confluence-specific document processing including
    content extraction, metadata handling, and exclusion configuration.
    """

    pass

class ConfluenceDatasourceManagerFactory(Factory):
    """Factory for creating Confluence datasource managers.

    This factory generates managers that handle the extraction of content from
    Confluence instances. It ensures proper configuration, reading, and parsing
    of Confluence content.

    Attributes:
        _configuration_class: Configuration class used for validating and processing
            Confluence-specific settings.
    """

    _configuration_class: Type = ConfluenceDatasourceConfiguration

    @classmethod
    def _create_instance(
        cls, configuration: ConfluenceDatasourceConfiguration
    ) -> BasicDatasourceManager:
        """Create a configured Confluence datasource manager.

        Sets up the necessary reader and parser components based on the provided
        configuration and assembles them into a functional manager.

        Args:
            configuration: Configuration object containing Confluence-specific
                parameters including authentication details, spaces to extract,
                and other extraction options.

        Returns:
            A fully initialized datasource manager that can extract and process
            data from Confluence.
        """
        reader = ConfluenceDatasourceReaderFactory.create(configuration)
        parser = ConfluenceDatasourceParserFactory.create(configuration)
        return BasicDatasourceManager(configuration, reader, parser)

class ConfluenceDatasourceParser(BaseParser[ConfluenceDocument]):

    def __init__(
        self,
        configuration: ConfluenceDatasourceConfiguration,
        parser: MarkItDown = MarkItDown(),
    ):
        """Initialize the Confluence parser with the provided configuration.

        Args:
            configuration: Configuration object containing Confluence connection details
        """
        self.configuration = configuration
        self.parser = parser

    def parse(self, page: str) -> ConfluenceDocument:
        """Parse a Confluence page into a document.

        Args:
            page: Dictionary containing Confluence page information

        Returns:
            ConfluenceDocument: Parsed document with extracted text and metadata
        """
        markdown = self._get_page_markdown(page)
        metadata = self._extract_metadata(page, self.configuration.base_url)
        return ConfluenceDocument(text=markdown, metadata=metadata)

    def _get_page_markdown(self, page: dict) -> str:
        """Extract markdown content from a Confluence page. Because of MarkItDown,
        we need to write the HTML content to a temporary file and then convert it to markdown.

        Args:
            page: Dictionary containing Confluence page details

        Returns:
            str: Markdown content of the page
        """
        html_content = page["body"]["view"]["value"]
        if not html_content:
            return ""

        with tempfile.NamedTemporaryFile(mode="w", suffix=".html") as temp_file:
            temp_file.write(html_content)
            temp_file.flush()
            return self.parser.convert(
                temp_file.name, file_extension=".html"
            ).text_content

    @staticmethod
    def _extract_metadata(page: dict, base_url: str) -> dict:
        """Extract and format page metadata.

        Args:
            page: Dictionary containing Confluence page details
            base_url: Base URL of the Confluence instance

        Returns:
            dict: Structured metadata including dates, IDs, and URLs
        """
        return {
            "created_time": page["history"]["createdDate"],
            "created_date": page["history"]["createdDate"].split("T")[0],
            "datasource": "confluence",
            "format": "md",
            "last_edited_date": page["history"]["lastUpdated"]["when"],
            "last_edited_time": page["history"]["lastUpdated"]["when"].split(
                "T"
            )[0],
            "page_id": page["id"],
            "space": page["_expandable"]["space"].split("/")[-1],
            "title": page["title"],
            "type": "page",
            "url": base_url + page["_links"]["webui"],
        }

class ConfluenceDatasourceParserFactory(Factory):
    _configuration_class: Type = ConfluenceDatasourceConfiguration

    @classmethod
    def _create_instance(
        cls, configuration: ConfluenceDatasourceConfiguration
    ) -> ConfluenceDatasourceParser:
        """Creates a Confluence parser instance.

        Args:
            configuration: Configuration object containing Confluence connection details

        Returns:
            ConfluenceDatasourceParser: Configured Confluence parser instance
        """
        return ConfluenceDatasourceParser(configuration)

class ConfluenceDatasourceReader(BaseReader):
    """Reader for extracting documents from Confluence spaces.

    Implements document extraction from Confluence spaces, handling pagination
    and export limits.
    """

    def __init__(
        self,
        configuration: ConfluenceDatasourceConfiguration,
        client: Confluence,
        logger: logging.Logger = LoggerConfiguration.get_logger(__name__),
    ):
        """Initialize the Confluence reader.

        Args:
            configuration: Settings for Confluence access and export limits
            client: Client for Confluence API interactions
            logger: Logger instance for recording operation information
        """
        super().__init__()
        self.export_limit = configuration.export_limit
        self.client = client
        self.logger = logger

    async def read_all_async(
        self,
    ) -> AsyncIterator[dict]:
        """Asynchronously fetch all documents from Confluence.

        Retrieves pages from all global spaces in Confluence, respecting the export limit.
        Yields each page as a dictionary containing its content and metadata.

        Returns:
            AsyncIterator[dict]: An async iterator of page dictionaries containing
            page content and metadata such as body, title, and last update information
        """
        self.logger.info(
            f"Fetching pages from Confluence with limit {self.export_limit}"
        )
        response = self.client.get_all_spaces(space_type="global")
        yield_counter = 0

        for space in response["results"]:
            space_limit = (
                self.export_limit - yield_counter
                if self.export_limit is not None
                else None
            )
            if space_limit is not None and space_limit <= 0:
                break

            space_pages = self._get_all_pages(space["key"], space_limit)
            for page in tqdm(
                space_pages,
                desc=f"[Confluence] Reading {space['key']} space pages content",
                unit="pages",
            ):
                yield_counter += 1
                if (
                    self.export_limit is not None
                    and yield_counter > self.export_limit
                ):
                    break
                yield page

    def _get_all_pages(self, space: str, limit: int) -> List[dict]:
        """Fetch all pages from a specific Confluence space.

        Handles pagination internally to retrieve all pages from the specified space,
        up to the optional limit. Pages include body content and update history.

        Args:
            space: Space key to fetch pages from
            limit: Maximum number of pages to fetch (None for unlimited)

        Returns:
            List[dict]: List of page dictionaries with content and metadata
        """
        start = 0
        params = {
            "space": space,
            "start": start,
            "status": None,
            "expand": "body.view,history.lastUpdated",
        }
        all_pages = []

        try:
            while True:
                pages = self.client.get_all_pages_from_space(**params)
                all_pages.extend(pages)

                if len(pages) == 0 or ConfluenceDatasourceReader._limit_reached(
                    all_pages, limit
                ):
                    break

                start = len(all_pages)
                params["start"] = start
        except HTTPError as e:
            self.logger.warning(f"Error while fetching pages from {space}: {e}")

        return all_pages if limit is None else all_pages[:limit]

    @staticmethod
    def _limit_reached(pages: List[dict], limit: int) -> bool:
        """Check if the page retrieval limit has been reached.

        Determines whether the number of fetched pages has reached or exceeded
        the specified limit.

        Args:
            pages: List of already retrieved pages
            limit: Maximum number of pages to retrieve (None for unlimited)

        Returns:
            bool: True if limit reached or exceeded, False otherwise
        """
        return limit is not None and len(pages) >= limit

class ConfluenceDatasourceReaderFactory(Factory):
    """Factory for creating Confluence reader instances.

    Creates and configures ConfluenceDatasourceReader objects with appropriate
    clients based on the provided configuration.
    """

    _configuration_class = ConfluenceDatasourceConfiguration

    @classmethod
    def _create_instance(
        cls, configuration: ConfluenceDatasourceConfiguration
    ) -> ConfluenceDatasourceReader:
        """Creates a configured Confluence reader instance.

        Initializes the Confluence client and reader with the given configuration
        settings for credentials, URL, and export limits.

        Args:
            configuration: Confluence connection and access settings

        Returns:
            ConfluenceDatasourceReader: Fully configured reader instance
        """
        client = ConfluenceClientFactory.create(configuration)
        return ConfluenceDatasourceReader(
            configuration=configuration,
            client=client,
        )