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
            parser: MarkItDown instance for converting HTML to markdown
        """
        self.configuration = configuration
        self.parser = parser

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

        Args:
            page: Confluence page details

        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: ConfluencePage) -> 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: 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: ConfluencePage, base_url: str) -> dict:
        """Extract and format page metadata.

        Args:
            page: 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[ConfluencePage]:
        """Asynchronously fetch all documents from Confluence.

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

        Returns:
            AsyncIterator[ConfluencePage]: An async iterator of Confluence pages.
        """
        self.logger.info(
            f"Reading pages from Confluence with limit {self.export_limit}"
        )
        response = self.client.get_all_spaces(space_type="global")
        spaces = [Space.model_validate(space) for space in response["results"]]
        yield_counter = 0

        for space in spaces:
            for page in self._get_all_pages(space.key):
                if self._limit_reached(yield_counter, self.export_limit):
                    return

                self.logger.info(
                    f"Fetched Confluence page {yield_counter}/{self.export_limit}."
                )
                yield_counter += 1
                yield page

    def _get_all_pages(self, space: str) -> Iterator[ConfluencePage]:
        """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:
            Iterator[ConfluencePage]: Iterator of Confluence pages with content and metadata
        """
        start = 0
        params = {
            "space": space,
            "start": start,
            "status": None,
            "expand": "body.view,history.lastUpdated",
        }

        try:
            while True:
                pages_raw = self.client.get_all_pages_from_space(**params)
                pages = [
                    ConfluencePage.model_validate(page) for page in pages_raw
                ]
                if not pages:
                    return

                for page in pages:
                    yield page

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

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,
        )

class ConfluencePage(BaseModel):
    """Model representing a Confluence page from the API."""

    id: str
    title: str
    body: Body
    history: History
    links: Links = Field(alias="_links")
    expandable: Dict[str, str] = Field(alias="_expandable")