Merge pull request #19 from epoch8/fix-drop-duplicates-on-update-from-ls

Drop Duplicates on fetch from LS + test

Author: elephantum

CHANGELOG.md

@@ -1,3 +1,6 @@
+# 0.3.1
+* Add drop_duplicates on fetch data from LabelStudio (get_annotations_from_ls)
+
 # 0.3.0
 
 * Update datapipe-core version (0.13.0-alpha.4)

datapipe_label_studio_lite/pipeline.py

@@ -383,6 +383,8 @@ def _cleanup(values):
                         ],
                     }
                 )
+                # Удаление возможных дубликатов из LabelStudio.
+                output_df = output_df.drop_duplicates(subset=self.primary_keys, keep="last")
                 output_dts[0].store_chunk(output_df)
 
             if len(updated_ats) > 0:

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "datapipe-label-studio-lite"
-version = "0.3.0"
+version = "0.3.1"
 description = ""
 authors = [
     "Alexander Kozlov <bobokvsky@epoch8.co>",

tests/test_pipeline.py

@@ -745,3 +745,56 @@ def _gen2():
         # Предсказания не должны уйти
         # if include_predictions:
         #     assert len(df_ls.loc[idx, 'predictions']) == include_prepredictions + include_predictions
+
+
+@parametrize_with_cases(
+    "ds, catalog, steps, project_title, include_preannotations, include_prepredictions, "
+    "include_predictions, label_studio_session, delete_unannotated_tasks_only_on_update",
+    cases=CasesLabelStudio,
+)
+def test_ls_moderation_with_duplicates_in_ls(
+    ds: DataStore,
+    catalog: Catalog,
+    steps: List[DatatableTransformStep],
+    project_title: str,
+    include_preannotations: bool,
+    include_prepredictions: bool,
+    include_predictions: bool,
+    label_studio_session: label_studio_sdk.Client,
+    delete_unannotated_tasks_only_on_update: bool,
+):
+    # This should be ok (project will be created, but without data)
+    run_steps(ds, steps)
+    run_steps(ds, steps)
+
+    # Загружаем данные для задач в LS во входную таблицу.
+    do_batch_generate(
+        func=gen_data_df,
+        ds=ds,
+        output_dts=[ds.get_table("ls_input_data_raw")],
+    )
+    
+    # Добавляем дубликаты задач напрямую в проект LS.
+    tasks_duplicates_to_add = [
+        {
+            "data": {
+                "id": "task_1",
+                "text": "task_1_new_text"
+            }
+        },
+        {
+            "data": {
+                "id": "task_2",
+                "text": "task_2_new_text"
+            }
+        }
+    ]
+    project = get_project_by_title(label_studio_session, project_title)
+    project.import_tasks(tasks=tasks_duplicates_to_add)
+
+    # Запускаем трансформацию.
+    run_steps(ds, steps)
+    
+    # Проверяем количество задач в LS и данных в выходной таблице трубы.
+    assert len(project.get_tasks()) == TASKS_COUNT + len(tasks_duplicates_to_add)
+    assert len(ds.get_table("ls_output").get_data()) == TASKS_COUNT