Consolidate docs

Author: pamelafox

app/backend/prepdocslib/blobmanager.py

@@ -432,6 +432,7 @@ async def upload_blob(self, file: File) -> str:
                 blob_client = await container_client.upload_blob(blob_name, reopened_file, overwrite=True)
                 file.url = blob_client.url
 
+        assert file.url is not None, "file.url must be set after upload"
         return unquote(file.url)
 
     async def upload_document_image(

app/backend/prepdocslib/cloudingestionstrategy.py

@@ -71,7 +71,7 @@ def __init__(
         use_acls: bool = False,
         use_multimodal: bool = False,
         enforce_access_control: bool = False,
-        search_user_assigned_identity_resource_id: str | None = None,
+        search_user_assigned_identity_resource_id: str,
     ) -> None:
         self.list_file_strategy = list_file_strategy
         self.blob_manager = blob_manager
@@ -297,38 +297,31 @@ async def setup(self) -> None:
             skillset = self._build_skillset()
             await indexer_client.create_or_update_skillset(skillset)
 
+            indexer = SearchIndexer(
+                name=self.indexer_name,
+                description="Indexer orchestrating cloud ingestion pipeline",
+                data_source_name=self.data_source_name,
+                target_index_name=self.search_info.index_name,
+                skillset_name=self.skillset_name,
+                parameters=IndexingParameters(
+                    configuration=IndexingParametersConfiguration(
+                        query_timeout=None,  # type: ignore
+                        data_to_extract="storageMetadata",
+                        allow_skillset_to_read_file_data=True,
+                    )
+                ),
+            )
+            await indexer_client.create_or_update_indexer(indexer)
+
     async def run(self) -> None:
-        if self.document_action == DocumentAction.Add:
-            files = self.list_file_strategy.list()
-            async for file in files:
-                try:
-                    await self.blob_manager.upload_blob(file)
-                finally:
-                    if file:
-                        file.close()
-        elif self.document_action == DocumentAction.Remove:
-            paths = self.list_file_strategy.list_paths()
-            async for path in paths:
-                await self.blob_manager.remove_blob(path)
-        elif self.document_action == DocumentAction.RemoveAll:
-            await self.blob_manager.remove_blob()
-
-        indexer = SearchIndexer(
-            name=self.indexer_name,
-            description="Indexer orchestrating cloud ingestion pipeline",
-            data_source_name=self.data_source_name,
-            target_index_name=self.search_info.index_name,
-            skillset_name=self.skillset_name,
-            parameters=IndexingParameters(
-                configuration=IndexingParametersConfiguration(
-                    query_timeout=None,
-                    data_to_extract="storageMetadata",
-                    allow_skillset_to_read_file_data=True,
-                )
-            ),
-        )
+        files = self.list_file_strategy.list()
+        async for file in files:
+            try:
+                await self.blob_manager.upload_blob(file)
+            finally:
+                if file:
+                    file.close()
 
         async with self.search_info.create_search_indexer_client() as indexer_client:
-            await indexer_client.create_or_update_indexer(indexer)
             await indexer_client.run_indexer(self.indexer_name)
         logger.info("Triggered indexer '%s' for cloud ingestion", self.indexer_name)

app/backend/prepdocslib/page.py

@@ -85,15 +85,25 @@ def from_skill_payload(cls, data: dict[str, Any]) -> tuple["ImageOnPage", str]:
         else:
             bbox = (0, 0, 0, 0)
 
+        filename = data.get("filename")
+        figure_id = data.get("figure_id")
+        placeholder = data.get("placeholder")
+        assert filename is not None, "filename is required"
+        assert figure_id is not None, "figure_id is required"
+
+        # Generate placeholder if not provided
+        if placeholder is None:
+            placeholder = f'<figure id="{figure_id}"></figure>'
+
         image = cls(
             bytes=raw_bytes,
             bbox=bbox,
             page_num=page_num,
-            filename=data.get("filename"),
-            figure_id=data.get("figure_id"),
-            placeholder=data.get("placeholder"),
+            filename=filename,
+            figure_id=figure_id,
+            placeholder=placeholder,
             mime_type=data.get("mime_type") or "image/png",
-            title=data.get("title"),
+            title=data.get("title") or "",
             description=data.get("description"),
             url=data.get("url"),
         )

app/backend/setup_cloud_ingestion.py

@@ -7,9 +7,11 @@
 
 from azure.core.credentials_async import AsyncTokenCredential
 from azure.identity.aio import AzureDeveloperCliCredential
+from openai import AsyncOpenAI
 from rich.logging import RichHandler
 
 from load_azd_env import load_azd_env
+from prepdocslib.blobmanager import BlobManager
 from prepdocslib.cloudingestionstrategy import CloudIngestionStrategy
 from prepdocslib.servicesetup import (
     OpenAIHost,
@@ -33,13 +35,13 @@ def clean_key_if_exists(key: Optional[str]) -> Optional[str]:
 async def setup_cloud_ingestion_strategy(
     azure_credential: AsyncTokenCredential,
     document_action: DocumentAction = DocumentAction.Add,
-) -> CloudIngestionStrategy:
+) -> tuple[CloudIngestionStrategy, AsyncOpenAI, AsyncTokenCredential, BlobManager]:
     """Setup the cloud ingestion strategy with all required services."""
 
     # Get environment variables
     search_service = os.environ["AZURE_SEARCH_SERVICE"]
     index_name = os.environ["AZURE_SEARCH_INDEX"]
-    search_user_assigned_identity_resource_id = os.environ.get("AZURE_SEARCH_USER_ASSIGNED_IDENTITY_RESOURCE_ID")
+    search_user_assigned_identity_resource_id = os.environ["AZURE_SEARCH_USER_ASSIGNED_IDENTITY_RESOURCE_ID"]
     storage_account = os.environ["AZURE_STORAGE_ACCOUNT"]
     storage_container = os.environ["AZURE_STORAGE_CONTAINER"]
     storage_resource_group = os.environ["AZURE_STORAGE_RESOURCE_GROUP"]
@@ -168,7 +170,8 @@ async def main():
         # Setup the indexer, skillset, and data source
         logger.info("Setting up indexer, skillset, and data source...")
         await ingestion_strategy.setup()
-        logger.info("Cloud ingestion setup complete!")
+        logger.info("Triggering initial indexing run...")
+        await ingestion_strategy.run()
 
     finally:
         # Gracefully close any async clients/credentials