chore: multi confluence support (#6)

* feat: Add CommaSeparatedStrList validation tests and update project versions

* feat: Enhance ConfluenceParameters model with additional confluence_kwargs field and clean up API generation script

* feat: Add asynchronous document enhancement method and adjust minimum page content length

* feat: Update ConfluenceParameters model and enhance validation methods for comma separated lists

* feat: Update ConfluenceParameters model to use model_dump_json for JSON representation

Author: a-klos

admin-api-lib/pyproject.toml

@@ -1,3 +1,14 @@
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"
+
+[tool.poetry]
+name = "admin-api-lib"
+version = "1.0.1"
+description = "The admin backend is responsible for the document management. This includes deletion, upload and returning the source document."
+authors = ["STACKIT Data and AI Consulting <[email protected]>"]
+packages = [{ include = "admin_api_lib", from = "src" }]
+
 [tool.flake8]
 exclude= [".eggs", "./rag-core-library/*", "./src/admin_api_lib/models/*", "./src/admin_api_lib/rag_backend_client/*", "./src/admin_api_lib/extractor_api_client/*", ".git", ".hg", ".mypy_cache", ".tox", ".venv", ".devcontainer", "venv", "_build", "buck-out", "build", "dist", "**/__init__.py"]
 statistics = true
@@ -18,7 +29,10 @@ per-file-ignores = """
   ./src/admin_api_lib/impl/admin_api.py: B008,
   ./src/admin_api_lib/dependency_container.py: CCE002,CCE001,
   ./src/admin_api_lib/apis/admin_api_base.py: WOT001,
-  ./tests/*: S101,
+  ./tests/*: S101,S106,D100,D103,PT011
+  ./src/admin_api_lib/impl/settings/confluence_settings.py: C901,N805,
+  ./src/admin_api_lib/impl/utils/comma_separated_bool_list.py: R505,
+  ./src/admin_api_lib/impl/utils/comma_separated_str_list.py: R505,
 """
 
 [tool.black]
@@ -51,12 +65,6 @@ known_local_folder = ["admin_api_lib", "rag_core_lib"]
 [tool.pylint]
 max-line-length = 120
 
-[tool.poetry]
-name = "admin_api_lib"
-version = "0.0.1"
-description = "The admin backend is responsible for the document management. This includes deletion, upload and returning the source document."
-authors = ["STACKIT Data and AI Consulting <[email protected]>"]
-
 [tool.poetry.group.dev.dependencies]
 debugpy = "^1.8.1"
 pytest = "^8.2.1"
@@ -85,11 +93,6 @@ black = "^23.9.1"
 # flake8-logging-format = "^2024.24.12"
 # flake8-docstrings = "^1.7.0"
 
-
-[build-system]
-requires = ["poetry-core"]
-build-backend = "poetry.core.masonry.api"
-
 [tool.poetry.dependencies]
 rag-core-lib = {path = "../rag-core-lib"}
 python = "^3.11"
@@ -103,3 +106,9 @@ tqdm = "^4.66.4"
 langfuse = "^2.39.1"
 redis = "^5.0.8"
 pyyaml = "^6.0.2"
+
+[tool.pytest.ini_options]
+log_cli = 1
+log_cli_level = "DEBUG"
+pythonpath = "src"
+testpaths = "src/tests"

admin-api-lib/src/admin_api_lib/dependency_container.py

@@ -148,7 +148,7 @@ class DependencyContainer(DeclarativeContainer):
     )
 
     summary_enhancer = List(
-        Singleton(PageSummaryEnhancer, summarizer),
+        Singleton(PageSummaryEnhancer, summarizer, chunker_settings),
     )
     untraced_information_enhancer = Singleton(
         GeneralEnhancer,

admin-api-lib/src/admin_api_lib/extractor_api_client/openapi_client/models/confluence_parameters.py

@@ -20,6 +20,9 @@
 from typing import Any, ClassVar, Dict, List, Optional, Set
 
 from pydantic import BaseModel, ConfigDict, Field, StrictBool, StrictStr
+from typing import Any, ClassVar, Dict, List, Optional
+from admin_api_lib.extractor_api_client.openapi_client.models.key_value_pair import KeyValuePair
+from typing import Optional, Set
 from typing_extensions import Self
 
 
@@ -43,6 +46,9 @@ class ConfluenceParameters(BaseModel):
     document_name: StrictStr = Field(
         description="The name that will be used to store the confluence db in the key value db and the vectordatabase (metadata.document)."
     )
+    confluence_kwargs: Optional[List[KeyValuePair]] = Field(
+        default=None, description="Additional kwargs like verify_ssl"
+    )
     __properties: ClassVar[List[str]] = [
         "url",
         "token",
@@ -51,6 +57,7 @@ class ConfluenceParameters(BaseModel):
         "keep_markdown_format",
         "keep_newlines",
         "document_name",
+        "confluence_kwargs",
     ]
 
     model_config = ConfigDict(
@@ -89,6 +96,13 @@ def to_dict(self) -> Dict[str, Any]:
             exclude=excluded_fields,
             exclude_none=True,
         )
+        # override the default output from pydantic by calling `to_dict()` of each item in confluence_kwargs (list)
+        _items = []
+        if self.confluence_kwargs:
+            for _item_confluence_kwargs in self.confluence_kwargs:
+                if _item_confluence_kwargs:
+                    _items.append(_item_confluence_kwargs.to_dict())
+            _dict["confluence_kwargs"] = _items
         return _dict
 
     @classmethod
@@ -113,6 +127,9 @@ def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional[Self]:
                 else True,
                 "keep_newlines": obj.get("keep_newlines") if obj.get("keep_newlines") is not None else True,
                 "document_name": obj.get("document_name"),
+                "confluence_kwargs": [KeyValuePair.from_dict(_item) for _item in obj["confluence_kwargs"]]
+                if obj.get("confluence_kwargs") is not None
+                else None,
             }
         )
         return _obj

admin-api-lib/src/admin_api_lib/impl/api_endpoints/default_confluence_loader.py

@@ -3,8 +3,10 @@
 import logging
 from asyncio import run
 from threading import Thread
+import threading
 
 from fastapi import HTTPException, status
+from langchain_core.documents import Document
 
 from admin_api_lib.api_endpoints.confluence_loader import ConfluenceLoader
 from admin_api_lib.api_endpoints.document_deleter import DocumentDeleter
@@ -81,7 +83,6 @@ def __init__(
         self._extractor_api = extractor_api
         self._rag_api = rag_api
         self._settings = settings
-        self._sanitize_document_name()
         self._key_value_store = key_value_store
         self._information_mapper = information_mapper
         self._information_enhancer = information_enhancer
@@ -100,10 +101,16 @@ async def aload_from_confluence(self) -> None:
         HTTPException
             If the Confluence loader is not configured or if a load is already in progress.
         """
-        if not (self._settings.url.strip() and self._settings.space_key.strip() and self._settings.token.strip()):
-            raise HTTPException(
-                status.HTTP_501_NOT_IMPLEMENTED, "The confluence loader is not configured! Required fields are missing."
-            )
+        for index in range(len(self._settings.url)):
+            if not (
+                self._settings.url[index].strip()
+                and self._settings.space_key[index].strip()
+                and self._settings.token[index].strip()
+            ):
+                raise HTTPException(
+                    status.HTTP_501_NOT_IMPLEMENTED,
+                    "The confluence loader is not configured! Required fields are missing.",
+                )
 
         if self._background_thread is not None and self._background_thread.is_alive():
             raise HTTPException(
@@ -113,51 +120,76 @@ async def aload_from_confluence(self) -> None:
         self._background_thread.start()
 
     async def _aload_from_confluence(self) -> None:
-        params = self._settings_mapper.map_settings_to_params(self._settings)
+        async def process_confluence(index):
+            logger.info("Loading from Confluence %s", self._settings.url[index])
+            self._sanitize_document_name(index=index)
+
+            params = self._settings_mapper.map_settings_to_params(self._settings, index)
+            try:
+                self._key_value_store.upsert(self._settings.document_name[index], Status.PROCESSING)
+                information_pieces = self._extractor_api.extract_from_confluence_post(params)
+                documents = [
+                    self._information_mapper.extractor_information_piece2document(x) for x in information_pieces
+                ]
+                documents = await self._aenhance_langchain_documents(documents)
+                chunked_documents = self._chunker.chunk(documents)
+                rag_information_pieces = [
+                    self._information_mapper.document2rag_information_piece(doc) for doc in chunked_documents
+                ]
+            except Exception as e:
+                self._key_value_store.upsert(self._settings.document_name[index], Status.ERROR)
+
+                logger.error("Error while loading from Confluence: %s", str(e))
+                raise HTTPException(
+                    status.HTTP_500_INTERNAL_SERVER_ERROR, f"Error loading from Confluence: {str(e)}"
+                ) from e
+
+            await self._delete_previous_information_pieces(index=index)
+            self._key_value_store.upsert(self._settings.document_name[index], Status.UPLOADING)
+            self._upload_information_pieces(rag_information_pieces, index=index)
+
+        threads = []
+        for idx in range(len(self._settings.url)):
+            t = threading.Thread(target=lambda idx=idx: run(process_confluence(idx)))
+            threads.append(t)
+            t.start()
+        for t in threads:
+            t.join()
+
+    async def _aenhance_langchain_documents(self, documents: list[Document]):
         try:
-            self._key_value_store.upsert(self._settings.document_name, Status.PROCESSING)
-            information_pieces = self._extractor_api.extract_from_confluence_post(params)
-            documents = [self._information_mapper.extractor_information_piece2document(x) for x in information_pieces]
-            chunked_documents = self._chunker.chunk(documents)
-            rag_information_pieces = [
-                self._information_mapper.document2rag_information_piece(doc) for doc in chunked_documents
-            ]
+            return await self._information_enhancer.ainvoke(documents)
         except Exception as e:
-            self._key_value_store.upsert(self._settings.document_name, Status.ERROR)
-            logger.error("Error while loading from Confluence: %s", str(e))
-            raise HTTPException(
-                status.HTTP_500_INTERNAL_SERVER_ERROR, f"Error loading from Confluence: {str(e)}"
-            ) from e
-
-        await self._delete_previous_information_pieces()
-        self._key_value_store.upsert(self._settings.document_name, Status.UPLOADING)
-        self._upload_information_pieces(rag_information_pieces)
+            logger.error("Exception occured while enhancing confluence langchain document %s" % e)
+            raise e
 
-    async def _delete_previous_information_pieces(self):
+    async def _delete_previous_information_pieces(self, index=0):
         try:
-            await self._document_deleter.adelete_document(self._settings.document_name)
+            await self._document_deleter.adelete_document(self._settings.document_name[index])
         except HTTPException as e:
             logger.error(
                 (
                     "Error while trying to delete documents with id: %s before uploading %s."
                     "NOTE: Still continuing with upload."
                 ),
-                self._settings.document_name,
+                self._settings.document_name[index],
                 e,
             )
 
-    def _upload_information_pieces(self, rag_api_documents):
+    def _upload_information_pieces(self, rag_api_documents, index=0):
         try:
             self._rag_api.upload_information_piece(rag_api_documents)
-            self._key_value_store.upsert(self._settings.document_name, Status.READY)
+            self._key_value_store.upsert(self._settings.document_name[index], Status.READY)
             logger.info("Confluence loaded successfully")
         except Exception as e:
-            self._key_value_store.upsert(self._settings.document_name, Status.ERROR)
+            self._key_value_store.upsert(self._settings.document_name[index], Status.ERROR)
             logger.error("Error while uploading Confluence to the database: %s", str(e))
             raise HTTPException(500, f"Error loading from Confluence: {str(e)}") from e
 
-    def _sanitize_document_name(self) -> None:
-        document_name = self._settings.document_name if self._settings.document_name else self._settings.url
+    def _sanitize_document_name(self, index) -> None:
+        document_name = (
+            self._settings.document_name[index] if self._settings.document_name[index] else self._settings.url[index]
+        )
         document_name = document_name.replace("http://", "").replace("https://", "")
 
-        self._settings.document_name = sanitize_document_name(document_name)
+        self._settings.document_name[index] = sanitize_document_name(document_name)

admin-api-lib/src/admin_api_lib/impl/information_enhancer/page_summary_enhancer.py

@@ -29,9 +29,15 @@ class PageSummaryEnhancer(SummaryEnhancer):
 
     async def _acreate_summary(self, information: list[Document], config: Optional[RunnableConfig]) -> list[Document]:
         # group infos by page, defaulting to page 1 if no page metadata
+        if self._chunker_settings:
+            filtered_information = [
+                info for info in information if len(info.page_content) > self._chunker_settings.max_size
+            ]
+        else:
+            filtered_information = information
         grouped = [
-            [info for info in information if info.metadata.get("page", self.DEFAULT_PAGE_NR) == page]
-            for page in {info_piece.metadata.get("page", self.DEFAULT_PAGE_NR) for info_piece in information}
+            [info for info in filtered_information if info.metadata.get("page", self.DEFAULT_PAGE_NR) == page]
+            for page in {info_piece.metadata.get("page", self.DEFAULT_PAGE_NR) for info_piece in filtered_information}
         ]
 
         summary_tasks = [self._asummarize_page(info_group, config) for info_group in tqdm(grouped)]

admin-api-lib/src/admin_api_lib/impl/information_enhancer/summary_enhancer.py

@@ -3,6 +3,7 @@
 from abc import abstractmethod
 from typing import Optional
 
+from admin_api_lib.impl.settings.chunker_settings import ChunkerSettings
 from langchain_core.documents import Document
 from langchain_core.runnables import RunnableConfig, ensure_config
 
@@ -26,7 +27,7 @@ class SummaryEnhancer(InformationEnhancer):
 
     INFORMATION_METADATA_TYPE = "type"
 
-    def __init__(self, summarizer: Summarizer):
+    def __init__(self, summarizer: Summarizer, chunker_settings: ChunkerSettings = None):
         """
         Initialize the SummaryEnhancer with a given Summarizer instance.
 
@@ -37,6 +38,7 @@ def __init__(self, summarizer: Summarizer):
         """
         super().__init__()
         self._summarizer = summarizer
+        self._chunker_settings = chunker_settings
 
     @staticmethod
     def _is_relevant(information: Document) -> bool:

admin-api-lib/src/admin_api_lib/impl/key_db/file_status_key_value_store.py

@@ -76,7 +76,6 @@ def upsert(self, file_name: str, file_status: Status) -> None:
         None
         """
         self.remove(file_name)
-
         self._redis.sadd(self.STORAGE_KEY, FileStatusKeyValueStore._to_str(file_name, file_status))
 
     def remove(self, file_name: str) -> None:

admin-api-lib/src/admin_api_lib/impl/mapper/confluence_settings_mapper.py

@@ -10,7 +10,7 @@ class ConfluenceSettingsMapper:
     """Mapper class for converting ConfluenceSettings to ConfluenceParameters."""
 
     @staticmethod
-    def map_settings_to_params(settings: ConfluenceSettings) -> ConfluenceParameters:
+    def map_settings_to_params(settings: ConfluenceSettings, index) -> ConfluenceParameters:
         """
         Map ConfluenceSettings to ConfluenceParameters.
 
@@ -25,11 +25,12 @@ def map_settings_to_params(settings: ConfluenceSettings) -> ConfluenceParameters
             The parameters object for API consumption.
         """
         return ConfluenceParameters(
-            url=settings.url,
-            token=settings.token,
-            space_key=settings.space_key,
-            include_attachments=settings.include_attachments,
-            keep_markdown_format=settings.keep_markdown_format,
-            keep_newlines=settings.keep_newlines,
-            document_name=settings.document_name,
+            url=settings.url[index],
+            token=settings.token[index],
+            space_key=settings.space_key[index],
+            include_attachments=settings.include_attachments[index],
+            keep_markdown_format=settings.keep_markdown_format[index],
+            keep_newlines=settings.keep_newlines[index],
+            document_name=settings.document_name[index],
+            confluence_kwargs=[{"key": "verify_ssl", "value": settings.verify_ssl[index]}],
         )