feat(LAB-4165): add audio asset import support

Author: paulruelle

src/kili/services/asset_import/__init__.py

@@ -7,6 +7,7 @@
     ImportValidationError,
 )
 
+from .audio import AudioDataImporter
 from .base import (
     BaseAbstractAssetImporter,
     LoggerParams,
@@ -24,6 +25,7 @@
     from kili.client import Kili
 
 importer_by_type: dict[str, type[BaseAbstractAssetImporter]] = {
+    "AUDIO": AudioDataImporter,
     "PDF": PdfDataImporter,
     "IMAGE": ImageDataImporter,
     "GEOSPATIAL": ImageDataImporter,

src/kili/services/asset_import/audio.py

@@ -0,0 +1,63 @@
+"""Functions to import assets into an AUDIO project."""
+
+import os
+from enum import Enum
+
+from kili.core.helpers import is_url
+from kili.domain.project import InputType
+
+from .base import (
+    BaseAbstractAssetImporter,
+    BatchParams,
+    ContentBatchImporter,
+)
+from .exceptions import ImportValidationError
+from .types import AssetLike
+
+
+class AudioDataType(Enum):
+    """Audio data type."""
+
+    LOCAL_FILE = "LOCAL_FILE"
+    HOSTED_FILE = "HOSTED_FILE"
+
+
+class AudioDataImporter(BaseAbstractAssetImporter):
+    """Class for importing data into an AUDIO project."""
+
+    @staticmethod
+    def get_data_type(assets: list[AssetLike]) -> AudioDataType:
+        """Determine the type of data to upload from the service payload."""
+        content_array = [asset.get("content", "") for asset in assets]
+        has_local_file = any(os.path.exists(content) for content in content_array)  # type: ignore
+        has_hosted_file = any(is_url(content) for content in content_array)
+        if has_local_file and has_hosted_file:
+            raise ImportValidationError(
+                """
+                Cannot upload hosted data and local files at the same time.
+                Please separate the assets into 2 calls
+                """
+            )
+        if has_local_file:
+            return AudioDataType.LOCAL_FILE
+        return AudioDataType.HOSTED_FILE
+
+    def import_assets(self, assets: list[AssetLike], input_type: InputType):
+        """Import AUDIO assets into Kili."""
+        self._check_upload_is_allowed(assets)
+        data_type = self.get_data_type(assets)
+        assets = self.filter_duplicate_external_ids(assets)
+        if data_type == AudioDataType.LOCAL_FILE:
+            assets = self.filter_local_assets(assets, self.raise_error)
+            batch_params = BatchParams(is_hosted=False, is_asynchronous=False)
+            batch_importer = ContentBatchImporter(
+                self.kili, self.project_params, batch_params, self.pbar
+            )
+        elif data_type == AudioDataType.HOSTED_FILE:
+            batch_params = BatchParams(is_hosted=True, is_asynchronous=False)
+            batch_importer = ContentBatchImporter(
+                self.kili, self.project_params, batch_params, self.pbar
+            )
+        else:
+            raise ImportValidationError
+        return self.import_assets_by_batch(assets, batch_importer)