feat: Enhances OneUp Health integration for document retrieval (M2-8883)

compose/minio/create_bucket.sh

@@ -7,11 +7,24 @@
 
 /usr/bin/mc config host add local http://minio:9000 minioaccess miniosecret;
 #/usr/bin/mc rm -r --force local/${CDN__BUCKET};
+#media bucket
 /usr/bin/mc mb -p local/${CDN__BUCKET};
 /usr/bin/mc policy set download local/${CDN__BUCKET};
 /usr/bin/mc policy set public local/${CDN__BUCKET};
 /usr/bin/mc anonymous set upload local/${CDN__BUCKET};
 /usr/bin/mc anonymous set download local/${CDN__BUCKET};
 /usr/bin/mc anonymous set public local/${CDN__BUCKET};
+#answer bucket
+/usr/bin/mc mb -p local/${CDN__BUCKET_ANSWER};
+/usr/bin/mc policy set download local/${CDN__BUCKET_ANSWER};
+/usr/bin/mc policy set public local/${CDN__BUCKET_ANSWER};
+/usr/bin/mc anonymous set upload local/${CDN__BUCKET_ANSWER};
+/usr/bin/mc anonymous set download local/${CDN__BUCKET_ANSWER};
+#operations bucket
+/usr/bin/mc mb -p local/${CDN__BUCKET_OPERATIONS};
+/usr/bin/mc policy set download local/${CDN__BUCKET_OPERATIONS};
+/usr/bin/mc policy set public local/${CDN__BUCKET_OPERATIONS};
+/usr/bin/mc anonymous set upload local/${CDN__BUCKET_OPERATIONS};
+/usr/bin/mc anonymous set download local/${CDN__BUCKET_OPERATIONS};
 
-exit 0;
+exit 0;

docker-compose.yaml

@@ -118,7 +118,9 @@ services:
     image: minio/mc
     container_name: mindlogger_minio_mc
     environment:
-      CDN__BUCKET: "${CDN__BUCKET:-media}"
+      CDN__BUCKET: "${CDN__BUCKET:-cmi-media-local}"
+      CDN__BUCKET_ANSWER: "${CDN__BUCKET_ANSWER:-cmi-answer-local}"
+      CDN__BUCKET_OPERATIONS: "${CDN__BUCKET_OPERATIONS:-cmi-ops-local}"
     depends_on:
       - minio
     volumes:

src/apps/answers/db/schemas.py

@@ -107,3 +107,4 @@ class AnswerEHRSchema(Base):
     activity_id = Column(UUID(as_uuid=True), index=True)
     ehr_storage_uri = Column(Text())
     ehr_ingestion_status = Column(Text())
+    meta = Column(JSONB())

src/apps/answers/domain/answers.py

@@ -18,6 +18,7 @@
 from apps.activity_flows.domain.flow_full import FlowFull, FlowHistoryWithActivityFlat, FlowHistoryWithActivityFull
 from apps.answers.domain.answer_items import AnswerItem, ItemAnswerCreate
 from apps.applets.domain.base import AppletBaseInfo
+from apps.integrations.oneup_health.service.domain import EHRMetadata
 from apps.integrations.prolific.domain import ProlificParamsActivityAnswer
 from apps.shared.domain import InternalModel, PublicModel, Response
 from apps.shared.domain.custom_validations import datetime_from_ms
@@ -748,6 +749,7 @@ class AnswerEHR(InternalModel):
     ehr_ingestion_status: EHRIngestionStatus
     activity_id: uuid.UUID
     ehr_storage_uri: str | None
+    meta: EHRMetadata | None
 
 
 class AnswerEHRFull(AnswerEHR):

src/apps/integrations/oneup_health/service/domain.py

@@ -1,5 +1,6 @@
 import uuid
 from datetime import datetime
+from enum import StrEnum
 
 from pydantic import Field
 
@@ -15,3 +16,19 @@ class EHRData(InternalModel):
     activity_id: uuid.UUID
     target_subject_id: uuid.UUID
     user_id: uuid.UUID
+
+
+class EHRFileTypeEnum(StrEnum):
+    DOCS = "DOCS"
+    EHR = "EHR"
+
+
+class EHRFileMetadata(InternalModel):
+    name: str
+    size: int
+    type: EHRFileTypeEnum
+
+
+class EHRMetadata(InternalModel):
+    zip_files: list[EHRFileMetadata]
+    storage_path: str

src/apps/integrations/oneup_health/service/ehr_storage.py

@@ -27,6 +27,19 @@ def ehr_zip_filename(data: EHRData) -> str:
         )
         return filename
 
+    @staticmethod
+    def docs_zip_filename(data: EHRData) -> str:
+        provider_name = (
+            slugify(data.healthcare_provider_name, separator="-")
+            if data.healthcare_provider_name
+            else data.healthcare_provider_id
+        )
+        filename = (
+            f"{data.target_subject_id}_{data.activity_id}_"
+            f"{data.submit_id}_{data.date.strftime('%Y%m%d')}_{provider_name}_DOCS.zip"
+        )
+        return filename
+
     def _get_storage_path(self, base_path: str, key: str) -> str:
         index = key.find(base_path)
         if index == -1:  # substring not found
@@ -59,7 +72,18 @@ async def upload_resources(self, data: EHRData) -> tuple[str, str]:
 
         return self._get_storage_path(base_path, key), filename
 
-    async def upload_ehr_zip(self, resources_files: list[str], data: EHRData) -> str:
+    async def upload_file(self, data: EHRData, filename: str, content: bytes) -> None:
+        base_path = self._get_base_path(data)
+        key = self._cdn_client.generate_key(FileScopeEnum.EHR, base_path, filename)
+
+        file_buffer = io.BytesIO(content)
+        file_buffer.seek(0)
+
+        await self._cdn_client.upload(key, file_buffer)
+
+        file_buffer.close()
+
+    async def upload_ehr_zip(self, resources_files: list[str], data: EHRData) -> tuple[str, int]:
         base_path = self._get_base_path(data)
         filename = EHRStorage.ehr_zip_filename(data)
         key = self._cdn_client.generate_key(FileScopeEnum.EHR, base_path, filename)
@@ -80,8 +104,9 @@ async def upload_ehr_zip(self, resources_files: list[str], data: EHRData) -> str
                     file_buffer.close()
 
             zip_buffer.seek(0)
+            file_size = zip_buffer.getbuffer().nbytes
             await self._cdn_client.upload(key, zip_buffer)
-            return key
+            return filename, file_size
         finally:
             zip_buffer.close()
 

src/apps/integrations/oneup_health/service/oneup_health.py

@@ -1,9 +1,14 @@
+import base64
 import hashlib
+import io
+import mimetypes
 import uuid
+import zipfile
 from datetime import datetime, timezone
 from functools import reduce
 
 import httpx
+from slugify import slugify
 
 from apps.integrations.oneup_health.errors import (
     OneUpHealthAPIError,
@@ -13,8 +18,8 @@
     OneUpHealthTokenExpiredError,
     OneUpHealthUserAlreadyExistsError,
 )
-from apps.integrations.oneup_health.service.domain import EHRData
-from apps.integrations.oneup_health.service.ehr_storage import create_ehr_storage
+from apps.integrations.oneup_health.service.domain import EHRData, EHRFileMetadata, EHRFileTypeEnum, EHRMetadata
+from apps.integrations.oneup_health.service.ehr_storage import EHRStorage, create_ehr_storage
 from apps.shared.exception import InternalServerError
 from config import settings
 
@@ -480,6 +485,83 @@ async def _get_resources(self, entry_url: str, oneup_user_id: int):
 
         return resources
 
+    def _get_document_references(self, resource_list) -> list[dict[str, str]]:
+        """Filter DocumentReference resources and extract their document URLs."""
+        doc_refs = [res for res in resource_list if res.get("resourceType") == "DocumentReference"]
+        documents = []
+        for doc_ref in doc_refs:
+            for content in doc_ref.get("content", []):
+                attachment = content.get("attachment", {})
+                title = attachment.get("title")
+                url = attachment.get("url")
+                if url:
+                    documents.append(dict(title=title, url=url))
+        return documents
+
+    def _get_extension_from_content_type(self, content_type: str) -> str | None:
+        """Return the file extension (with dot) for a given MIME content type, or None if unknown."""
+        # Handle common edge cases if needed
+        if content_type == "application/xml":
+            return ".xml"
+
+        if content_type == "application/json":
+            return ".json"
+
+        return mimetypes.guess_extension(content_type)
+
+    async def _download_and_store_documents(
+        self, ehr_storage, oneup_user_id: int, data: EHRData
+    ) -> tuple[str | None, int | None]:
+        document_references = self._get_document_references(data.resources)
+
+        if len(document_references) == 0:
+            logger.info(f"No documents found for activity_id {data.activity_id}, submit_id {data.submit_id}")
+            return None, None
+
+        zip_filename = EHRStorage.docs_zip_filename(data)
+
+        # TODO: Optimize this function to avoid loading all files in memory when creating the zip file.
+        # Current implementation loads all document content into memory before writing to the zip file,
+        # which can cause memory issues with large documents or many documents.
+        # Consider using a streaming approach or temporary files to reduce memory usage.
+        zip_buffer = io.BytesIO()
+        try:
+            with zipfile.ZipFile(zip_buffer, "w", compression=zipfile.ZIP_DEFLATED) as zip_file:
+                for reference in document_references:
+                    url = reference["url"]
+                    try:
+                        document_meta = await self._client.get(url, headers={"x-oneup-user-id": str(oneup_user_id)})
+                        title = reference.get("title")
+                        last_updated_str = document_meta.get("meta", {}).get("lastUpdated")
+                        date = (
+                            datetime.fromisoformat(last_updated_str.replace("Z", "+00:00"))
+                            if last_updated_str
+                            else None
+                        )
+                        if date:
+                            date_string = date.strftime("%Y%m%d_%H%M%S")
+                        ext = self._get_extension_from_content_type(document_meta.get("contentType"))
+                        logger.info(f"Guessed extension: {ext} for mime type: {document_meta.get('contentType')}")
+                        file_name = f"{slugify(title) if title else url.split('/')[-1]}_{date_string}{ext}"
+                        data_b64: str = document_meta.get("data")
+                        content = base64.b64decode(data_b64)
+                        await ehr_storage.upload_file(data, file_name, content)
+                        zip_file.writestr(file_name, content)
+                        logger.info(f"Downloaded and stored document: {file_name}")
+
+                    except OneUpHealthAPIError as ex:
+                        logger.error(f"Failed to download document: {url}: {ex}")
+                        continue
+
+            zip_buffer.seek(0)
+            zip_size = zip_buffer.getbuffer().nbytes
+            await ehr_storage.upload_file(data, zip_filename, zip_buffer.getvalue())
+
+        finally:
+            zip_buffer.close()
+
+        return zip_filename, zip_size
+
     async def retrieve_patient_data(
         self,
         session,
@@ -490,7 +572,7 @@ async def retrieve_patient_data(
         activity_id: uuid.UUID,
         oneup_user_id: int,
         healthcare_providers: list[dict[str, str]],
-    ) -> str | None:
+    ) -> EHRMetadata | None:
         """
         Retrieve and store patient data for a subject.
 
@@ -522,28 +604,28 @@ async def retrieve_patient_data(
         entries = result.get("entry", [])
         ehr_storage = await create_ehr_storage(session=session, applet_id=applet_id)
         resource_files = []
+        zip_files = []
         for entry in entries:
             resource_url = entry.get("fullUrl")
             if resource_url:
                 logger.info(f"Retrieving resources from {resource_url}")
                 resources = await self._get_resources(f"{resource_url}/$everything?_count=100", oneup_user_id)
                 if len(resources) > 0:
-                    # Get the healthcare provider name
-                    healthcare_provider_id = entry.get("resource", {}).get("id")
+                    # Get the healthcare provider meta data
                     meta_source = entry.get("resource", {}).get("meta", {}).get("source")
-                    healthcare_provider_name = next(
+                    healthcare_provider = next(
                         (
-                            healthcare_provider["name"]
+                            healthcare_provider
                             for healthcare_provider in healthcare_providers
                             if f"1up-external-system:{healthcare_provider['id']}" == meta_source
                         ),
-                        None,
+                        {},
                     )
 
                     data = EHRData(
                         resources=resources,
-                        healthcare_provider_id=healthcare_provider_id,
-                        healthcare_provider_name=healthcare_provider_name,
+                        healthcare_provider_id=healthcare_provider.get("id"),
+                        healthcare_provider_name=healthcare_provider.get("name"),
                         date=datetime.now(timezone.utc),
                         submit_id=submit_id,
                         activity_id=activity_id,
@@ -555,8 +637,19 @@ async def retrieve_patient_data(
                     resource_files.append(f"{storage_path}/{filename}")
                     logger.info(
                         f"Stored EHR data for healthcare provider "
-                        f"{healthcare_provider_name} in {storage_path}/{filename}"
+                        f"{healthcare_provider.get('name')} in {storage_path}/{filename}"
                     )
+                    docs_zip_filename, docs_zip_size = await self._download_and_store_documents(
+                        ehr_storage, oneup_user_id, data
+                    )
+                    if docs_zip_filename:
+                        zip_files.append(
+                            EHRFileMetadata(name=docs_zip_filename, size=docs_zip_size, type=EHRFileTypeEnum.DOCS)
+                        )
+                        logger.info(
+                            f"Stored documents for healthcare provider {healthcare_provider.get('name')} "
+                            f"in {docs_zip_filename} size {docs_zip_size}"
+                        )
 
         # Upload EHR zip with all resources
         data = EHRData(
@@ -566,8 +659,12 @@ async def retrieve_patient_data(
             target_subject_id=target_subject_id,
             user_id=user_id,
         )
-        zip_file_path = await ehr_storage.upload_ehr_zip(resource_files, data)
+        ehr_zip_filename, ehr_zip_size = await ehr_storage.upload_ehr_zip(resource_files, data)
+        zip_files.append(EHRFileMetadata(name=ehr_zip_filename, size=ehr_zip_size, type=EHRFileTypeEnum.EHR))
 
-        logger.info(f"Stored EHR data in {zip_file_path}")
+        logger.info(f"Stored EHR data in {ehr_zip_filename} size {ehr_zip_size}")
 
-        return storage_path
+        return EHRMetadata(
+            zip_files=zip_files,
+            storage_path=storage_path,
+        )

src/apps/integrations/oneup_health/service/task.py

@@ -10,6 +10,7 @@
 from apps.answers.crud.answers import AnswersEHRCRUD
 from apps.answers.deps.preprocess_arbitrary import get_answer_session, preprocess_arbitrary_url
 from apps.answers.domain import AnswerEHR, EHRIngestionStatus
+from apps.integrations.oneup_health.service.domain import EHRMetadata
 from apps.integrations.oneup_health.service.oneup_health import OneupHealthService
 from apps.shared.exception import BaseError
 from broker import broker
@@ -57,7 +58,7 @@ async def _process_data_transfer(
     activity_id: uuid.UUID,
     oneup_user_id: int,
     start_date: datetime | None,
-) -> str | None:
+) -> EHRMetadata | None:
     """
     Process the OneUp Health data transfer for a subject.
 
@@ -208,7 +209,7 @@ async def task_ingest_user_data(
     Returns:
         list | None: List of retrieved resources if successful, None otherwise
     """
-    storage_path = None
+    ehr_metadata = None
 
     async with session_manager.get_session()() as session:
         info = await preprocess_arbitrary_url(applet_id=applet_id, session=session)
@@ -239,7 +240,7 @@ async def task_ingest_user_data(
 
                     # Process data transfer
 
-                    storage_path = await _process_data_transfer(
+                    ehr_metadata = await _process_data_transfer(
                         session=session,
                         user_id=user_id,
                         target_subject_id=target_subject_id,
@@ -249,7 +250,7 @@ async def task_ingest_user_data(
                         oneup_user_id=oneup_user_id,
                         start_date=start_date,
                     )
-                    if storage_path is None:
+                    if ehr_metadata is None:
                         logger.info(f"Data transfer not complete for OneUp Health user ID {oneup_user_id}")
                         # Error retry count is reset to default 0 if we are not in an error state.
                         to_reschedule = await _schedule_retry(
@@ -278,7 +279,8 @@ async def task_ingest_user_data(
                             submit_id=submit_id,
                             ehr_ingestion_status=EHRIngestionStatus.COMPLETED,
                             activity_id=activity_id,
-                            ehr_storage_uri=storage_path,
+                            ehr_storage_uri=ehr_metadata.storage_path,
+                            meta=ehr_metadata,
                         )
                     )
             except (BaseError, httpx.RequestError) as e:
@@ -300,4 +302,4 @@ async def task_ingest_user_data(
                     failed_attempts,
                 )
 
-            return storage_path
+            return ehr_metadata.storage_path if ehr_metadata is not None else None