added new CopyMoveItems usecase

Author: Narek Mkhitaryan

src/superannotate/lib/app/interface/sdk_interface.py

@@ -2318,7 +2318,7 @@ def invite_contributors_to_team(
 
     def get_annotations(
         self,
-        project: Union[int, NotEmptyStr],
+        project: Union[NotEmptyStr, int],
         items: Optional[Union[List[NotEmptyStr], List[int]]] = None,
     ):
         """Returns annotations for the given list of items.
@@ -2928,21 +2928,35 @@ def copy_items(
         destination: Union[NotEmptyStr, dict],
         items: Optional[List[NotEmptyStr]] = None,
         include_annotations: Optional[bool] = True,
+        duplicate_strategy: Literal[
+            "skip", "replace", "replace_annotations_only"
+        ] = "skip",
     ):
         """Copy images in bulk between folders in a project
 
-        :param source: project name or folder path to select items from (e.g., “project1/folder1”).
+        :param source: project name (root) or folder path to pick items from (e.g., “project1/folder1”).
         :type source: str
 
-        :param destination: project name (root) or folder path to place copied items.
+        :param destination: project name (root) or folder path to place copied items (e.g., “project1/folder2”).
         :type destination: str
 
         :param items: names of items to copy. If None, all items from the source directory will be copied.
         :type items: list of str
 
-        :param include_annotations: enables annotations copy
+        :param include_annotations: enables the copying of item data, including annotations, status, priority score,
+         approval state, and category. If set to False, only the items will be copied without additional data.
         :type include_annotations: bool
 
+        :param duplicate_strategy: Specifies the strategy for handling duplicate items in the destination.
+         The default value is "skip".
+
+            - "skip": skips duplicate items in the destination and continues with the next item.
+            - "replace": replaces the annotations, status, priority score, approval state, and category of duplicate items.
+            - "replace_annotations_only": replaces only the annotations of duplicate items,
+              leaving other data (status, priority score, approval state, and category) unchanged.
+
+        :type duplicate_strategy: Literal["skip", "replace", "replace_annotations_only"]
+
         :return: list of skipped item names
         :rtype: list of strs
         """
@@ -2960,6 +2974,7 @@ def copy_items(
             to_folder=to_folder,
             item_names=items,
             include_annotations=include_annotations,
+            duplicate_strategy=duplicate_strategy,
         )
         if response.errors:
             raise AppException(response.errors)
@@ -2971,18 +2986,31 @@ def move_items(
         source: Union[NotEmptyStr, dict],
         destination: Union[NotEmptyStr, dict],
         items: Optional[List[NotEmptyStr]] = None,
+        duplicate_strategy: Literal[
+            "skip", "replace", "replace_annotations_only"
+        ] = "skip",
     ):
         """Move images in bulk between folders in a project
 
-        :param source: project name or folder path to pick items from (e.g., “project1/folder1”).
+        :param source: project name (root) or folder path to pick items from (e.g., “project1/folder1”).
         :type source: str
 
-        :param destination: project name (root) or folder path to move items to.
+        :param destination: project name (root) or folder path to move items to (e.g., “project1/folder2”).
         :type destination: str
 
         :param items: names of items to move. If None, all items from the source directory will be moved.
         :type items: list of str
 
+        :param duplicate_strategy: Specifies the strategy for handling duplicate items in the destination.
+         The default value is "skip".
+
+            - "skip": skips duplicate items in the destination and continues with the next item.
+            - "replace": replaces the annotations, status, priority score, approval state, and category of duplicate items.
+            - "replace_annotations_only": replaces only the annotations of duplicate items,
+              leaving other data (status, priority score, approval state, and category) unchanged.
+
+        :type duplicate_strategy: Literal["skip", "replace", "replace_annotations_only"]
+
         :return: list of skipped item names
         :rtype: list of strs
         """
@@ -3000,6 +3028,7 @@ def move_items(
             from_folder=source_folder,
             to_folder=destination_folder,
             item_names=items,
+            duplicate_strategy=duplicate_strategy,
         )
         if response.errors:
             raise AppException(response.errors)

src/superannotate/lib/core/serviceproviders.py

@@ -5,6 +5,7 @@
 from typing import Callable
 from typing import Dict
 from typing import List
+from typing import Literal
 
 from lib.core import entities
 from lib.core.conditions import Condition
@@ -287,10 +288,30 @@ def copy_multiple(
     ) -> ServiceResponse:
         raise NotImplementedError
 
+    @abstractmethod
+    def copy_move_multiple(
+        self,
+        project: entities.ProjectEntity,
+        from_folder: entities.FolderEntity,
+        to_folder: entities.FolderEntity,
+        item_names: List[str],
+        duplicate_strategy: Literal["skip", "replace", "replace_annotations_only"],
+        operation: Literal["copy", "move"],
+        include_annotations: bool = False,
+        include_pin: bool = False,
+    ) -> ServiceResponse:
+        raise NotImplementedError
+
     @abstractmethod
     def await_copy(self, project: entities.ProjectEntity, poll_id: int, items_count):
         raise NotImplementedError
 
+    @abstractmethod
+    def await_copy_move(
+        self, project: entities.ProjectEntity, poll_id: int, items_count
+    ):
+        raise NotImplementedError
+
     @abstractmethod
     def set_statuses(
         self,

src/superannotate/lib/core/usecases/items.py

@@ -33,6 +33,7 @@
 from lib.core.usecases.folders import SearchFoldersUseCase
 from lib.infrastructure.utils import divide_to_chunks
 from lib.infrastructure.utils import extract_project_folder
+from typing_extensions import Literal
 
 logger = logging.getLogger("sa")
 
@@ -537,6 +538,154 @@ def execute(self):
         return self._response
 
 
+class CopyMoveItems(BaseReportableUseCase):
+    """
+    Copy/Move items in bulk between folders in a project.
+    Return skipped item names.
+    """
+
+    def __init__(
+        self,
+        reporter: Reporter,
+        project: ProjectEntity,
+        from_folder: FolderEntity,
+        to_folder: FolderEntity,
+        item_names: List[str],
+        service_provider: BaseServiceProvider,
+        include_annotations: bool,
+        duplicate_strategy: Literal["skip", "replace", "replace_annotations_only"],
+        operation: Literal["copy", "move"],
+        chunk_size: int = 1000,
+    ):
+        super().__init__(reporter)
+        self._project = project
+        self._from_folder = from_folder
+        self._to_folder = to_folder
+        self._item_names = item_names
+        self._service_provider = service_provider
+        self._include_annotations = include_annotations
+        self._duplicate_strategy = duplicate_strategy
+        self._operation = operation
+        self._chunk_size = chunk_size
+
+    def _validate_limitations(self, items_count):
+        response = self._service_provider.get_limitations(
+            project=self._project,
+            folder=self._to_folder,
+        )
+        if not response.ok:
+            raise AppValidationException(response.error)
+        if self._operation == "copy":
+            folder_limit_err_msg = constants.COPY_FOLDER_LIMIT_ERROR_MESSAGE
+            project_limit_err_msg = constants.COPY_PROJECT_LIMIT_ERROR_MESSAGE
+        else:
+            folder_limit_err_msg = constants.MOVE_FOLDER_LIMIT_ERROR_MESSAGE
+            project_limit_err_msg = constants.MOVE_PROJECT_LIMIT_ERROR_MESSAGE
+        if items_count > response.data.folder_limit.remaining_image_count:
+            raise AppValidationException(folder_limit_err_msg)
+        if items_count > response.data.project_limit.remaining_image_count:
+            raise AppValidationException(project_limit_err_msg)
+
+    def validate_item_names(self):
+        if self._item_names:
+            self._item_names = list(set(self._item_names))
+
+    def execute(self):
+        if self.is_valid():
+            if self._item_names:
+                items = self._item_names
+            else:
+                res = self._service_provider.item_service.list(
+                    self._project.id, self._from_folder.id, EmptyQuery()
+                )
+                if res.error:
+                    raise AppException(res.error)
+                items = [i.name for i in res.data]
+            try:
+                self._validate_limitations(len(items))
+            except AppValidationException as e:
+                self._response.errors = e
+                return self._response
+            skipped_items = []
+            if self._duplicate_strategy == "skip":
+                existing_items = []
+                for i in range(0, len(items), self._chunk_size):
+                    query = Filter(
+                        "name", items[i : i + self._chunk_size], OperatorEnum.IN
+                    )  # noqa
+                    res = self._service_provider.item_service.list(
+                        self._project.id, self._to_folder.id, query
+                    )
+                    if res.error:
+                        raise AppException(res.error)
+                    if not res.data:
+                        continue
+                    existing_items += res.data
+                duplications = [item.name for item in existing_items]
+                items_to_processing = list(set(items) - set(duplications))
+                skipped_items.extend(duplications)
+            else:
+                items_to_processing = items
+            if items_to_processing:
+                for i in range(0, len(items_to_processing), self._chunk_size):
+                    chunk_to_process = items_to_processing[
+                        i : i + self._chunk_size
+                    ]  # noqa: E203
+                    response = self._service_provider.items.copy_move_multiple(
+                        project=self._project,
+                        from_folder=self._from_folder,
+                        to_folder=self._to_folder,
+                        item_names=chunk_to_process,
+                        include_annotations=self._include_annotations,
+                        duplicate_strategy=self._duplicate_strategy,
+                        operation=self._operation,
+                    )
+                    if not response.ok or not response.data.get("poll_id"):
+                        skipped_items.extend(chunk_to_process)
+                        continue
+                    try:
+                        self._service_provider.items.await_copy_move(
+                            project=self._project,
+                            poll_id=response.data["poll_id"],
+                            items_count=len(chunk_to_process),
+                        )
+                    except BackendError as e:
+                        self._response.errors = AppException(e)
+                        return self._response
+                existing_items = []
+                for i in range(0, len(items_to_processing), self._chunk_size):
+                    res = self._service_provider.item_service.list(
+                        self._project.id,
+                        self._to_folder.id,
+                        Filter(
+                            "name",
+                            items_to_processing[i : i + self._chunk_size],
+                            OperatorEnum.IN,
+                        ),  # noqa
+                    )
+                    if res.error:
+                        raise AppException(res.error)
+
+                    existing_items += res.data
+
+                existing_item_names_set = {item.name for item in existing_items}
+                items_to_processing_names_set = set(items_to_processing)
+                processed_items = existing_item_names_set.intersection(
+                    items_to_processing_names_set
+                )
+                skipped_items.extend(
+                    list(items_to_processing_names_set - processed_items)
+                )
+                operation_processing_map = {"copy": "Copied", "move": "Moved"}
+                self.reporter.log_info(
+                    f"{operation_processing_map[self._operation]} {len(processed_items)}/{len(items_to_processing)} item(s) from "
+                    f"{self._project.name}{'' if self._from_folder.is_root else f'/{self._from_folder.name}'} to "
+                    f"{self._project.name}{'' if self._to_folder.is_root else f'/{self._to_folder.name}'}"
+                )
+            self._response.data = list(set(skipped_items))
+        return self._response
+
+
 class SetAnnotationStatues(BaseReportableUseCase):
     CHUNK_SIZE = 500
     ERROR_MESSAGE = "Failed to change status"

src/superannotate/lib/infrastructure/controller.py

@@ -6,6 +6,7 @@
 from typing import Callable
 from typing import Dict
 from typing import List
+from typing import Literal
 from typing import Optional
 from typing import Tuple
 from typing import Union
@@ -28,6 +29,7 @@
 from lib.core.entities import UserEntity
 from lib.core.entities.classes import AnnotationClassEntity
 from lib.core.entities.integrations import IntegrationEntity
+from lib.core.enums import ProjectType
 from lib.core.exceptions import AppException
 from lib.core.jsx_conditions import EmptyQuery
 from lib.core.jsx_conditions import Filter
@@ -540,35 +542,65 @@ def copy_multiple(
         project: ProjectEntity,
         from_folder: FolderEntity,
         to_folder: FolderEntity,
+        duplicate_strategy: Literal["skip", "replace", "replace_annotations_only"],
         item_names: List[str] = None,
         include_annotations: bool = False,
     ):
-        use_case = usecases.CopyItems(
-            reporter=Reporter(),
-            project=project,
-            from_folder=from_folder,
-            to_folder=to_folder,
-            item_names=item_names,
-            service_provider=self.service_provider,
-            include_annotations=include_annotations,
-        )
+        if project.type == ProjectType.PIXEL:
+            use_case = usecases.CopyItems(
+                reporter=Reporter(),
+                project=project,
+                from_folder=from_folder,
+                to_folder=to_folder,
+                item_names=item_names,
+                service_provider=self.service_provider,
+                include_annotations=include_annotations,
+            )
+        else:
+            use_case = usecases.CopyMoveItems(
+                reporter=Reporter(),
+                project=project,
+                from_folder=from_folder,
+                to_folder=to_folder,
+                item_names=item_names,
+                service_provider=self.service_provider,
+                include_annotations=include_annotations,
+                duplicate_strategy=duplicate_strategy,
+                operation="copy",
+                chunk_size=500,
+            )
         return use_case.execute()
 
     def move_multiple(
         self,
         project: ProjectEntity,
         from_folder: FolderEntity,
         to_folder: FolderEntity,
+        duplicate_strategy: Literal["skip", "replace", "replace_annotations_only"],
         item_names: List[str] = None,
     ):
-        use_case = usecases.MoveItems(
-            reporter=Reporter(),
-            project=project,
-            from_folder=from_folder,
-            to_folder=to_folder,
-            item_names=item_names,
-            service_provider=self.service_provider,
-        )
+        if project.type == ProjectType.PIXEL:
+            use_case = usecases.MoveItems(
+                reporter=Reporter(),
+                project=project,
+                from_folder=from_folder,
+                to_folder=to_folder,
+                item_names=item_names,
+                service_provider=self.service_provider,
+            )
+        else:
+            use_case = usecases.CopyMoveItems(
+                reporter=Reporter(),
+                project=project,
+                from_folder=from_folder,
+                to_folder=to_folder,
+                item_names=item_names,
+                service_provider=self.service_provider,
+                duplicate_strategy=duplicate_strategy,
+                include_annotations=True,
+                operation="move",
+                chunk_size=1000,
+            )
         return use_case.execute()
 
     def set_annotation_statuses(