[utils]受入/検査がスキップされたタスク履歴番号を取得するメソッドを追加

annofabapi/utils.py

@@ -7,12 +7,14 @@
 import datetime
 import logging
 from pathlib import Path
-from typing import Optional
+from typing import List, Optional
 
 import dateutil
 import dateutil.tz
 import requests
 
+from annofabapi.models import TaskHistory, TaskPhase
+
 
 def raise_for_status(response: requests.Response):
     """
@@ -131,3 +133,61 @@ def wrapped(*args, **kwargs):
             logging.getLogger("backoff").setLevel(level=backoff_logger_level)
 
     return wrapped
+
+
+def get_task_history_index_skipped_acceptance(task_history_list: List[TaskHistory]) -> List[int]:
+    """
+    受入がスキップされたタスク履歴のインデックス番号（0始まり）を返す。
+​
+    Args:
+        task_history_list: タスク履歴List
+​
+    Returns:
+        受入フェーズがスキップされた履歴のインデックス番号（0始まり）。受入がスキップされていない場合は空リストを返す。
+​
+    """
+    index_list = []
+    for index, history in enumerate(task_history_list):
+        if not (TaskPhase(history["phase"]) == TaskPhase.ACCEPTANCE and history["account_id"] is None):
+            continue
+
+        if index + 1 < len(task_history_list):
+            # 直後の履歴あり
+            next_history = task_history_list[index + 1]
+            # 直後の履歴が受入の場合、受入がスキップされた履歴（提出取り消しではない）
+            if TaskPhase(next_history["phase"]) == TaskPhase.ACCEPTANCE:
+                index_list.append(index)
+        else:
+            # 直後の履歴がない
+            index_list.append(index)
+
+    return index_list
+
+
+def get_task_history_index_skipped_inspection(task_history_list: List[TaskHistory]) -> List[int]:
+    """
+    検査フェーズがスキップされたタスク履歴のインデックス番号（0始まり）を返す。
+​
+    Args:
+        task_history_list: タスク履歴List
+​
+    Returns:
+        検査フェーズがスキップされた履歴のインデックス番号（0始まり）。検査がスキップされていない場合は空リストを返す。
+​
+    """
+    index_list = []
+    for index, history in enumerate(task_history_list):
+        if not (TaskPhase(history["phase"]) == TaskPhase.INSPECTION and history["account_id"] is None):
+            continue
+
+        if index + 1 < len(task_history_list):
+            # 直後の履歴あり
+            next_history = task_history_list[index + 1]
+            # 直後の履歴が受入/検査の場合、検査がスキップされた履歴（提出取り消しではない）
+            if TaskPhase(next_history["phase"]) in [TaskPhase.ACCEPTANCE, TaskPhase.INSPECTION]:
+                index_list.append(index)
+        else:
+            # 直後の履歴がない
+            index_list.append(index)
+
+    return index_list

tests/test_local_utils.py

@@ -3,3 +3,79 @@
 #     d = datetime.datetime(2019, 10, 8, 16, 20, 8, 241762)
 #     tz_jst = datetime.timezone(datetime.timedelta(hours=9))
 #     assert to_iso8601_extension(d, tz_jst) == "2019-10-08T16:20:08.241+09:00"
+
+from annofabapi.utils import get_task_history_index_skipped_acceptance, get_task_history_index_skipped_inspection
+
+
+class TestTaskHistoryUtils:
+    ACCOUNT_ID = "12345678-abcd-1234-abcd-1234abcd5678"
+
+    def test_get_task_history_index_skipped_acceptance(self):
+        task_history_list = [{
+            "phase": "annotation",
+            "phase_stage": 1,
+            "account_id": self.ACCOUNT_ID
+        }, {
+            "phase": "acceptance",
+            "phase_stage": 1,
+            "account_id": None
+        }]
+
+        actual = get_task_history_index_skipped_acceptance(task_history_list)
+        expected = [1]
+        assert all([a == b for a, b in zip(actual, expected)])
+
+    def test_get_task_history_index_skipped_acceptance_提出取消は対象外(self):
+        task_history_list = [{
+            "phase": "annotation",
+            "phase_stage": 1,
+            "account_id": self.ACCOUNT_ID
+        }, {
+            "phase": "acceptance",
+            "phase_stage": 1,
+            "account_id": None
+        }, {
+            "phase": "annotation",
+            "phase_stage": 1,
+            "account_id": self.ACCOUNT_ID
+        }]
+        actual = get_task_history_index_skipped_acceptance(task_history_list)
+        expected = []
+        assert all([a == b for a, b in zip(actual, expected)])
+
+    def test_get_task_history_index_skipped_inspection(self):
+        task_history_list = [{
+            "phase": "annotation",
+            "phase_stage": 1,
+            "account_id": self.ACCOUNT_ID
+        }, {
+            "phase": "inspection",
+            "phase_stage": 1,
+            "account_id": None
+        }, {
+            "phase": "acceptance",
+            "phase_stage": 1,
+            "account_id": None
+        }]
+
+        actual = get_task_history_index_skipped_inspection(task_history_list)
+        expected = [1]
+        assert all([a == b for a, b in zip(actual, expected)])
+
+    def test_get_task_history_index_skipped_inspection_提出取消は対象外(self):
+        task_history_list = [{
+            "phase": "annotation",
+            "phase_stage": 1,
+            "account_id": self.ACCOUNT_ID
+        }, {
+            "phase": "inspection",
+            "phase_stage": 1,
+            "account_id": None
+        }, {
+            "phase": "annotation",
+            "phase_stage": 1,
+            "account_id": self.ACCOUNT_ID
+        }]
+        actual = get_task_history_index_skipped_inspection(task_history_list)
+        expected = []
+        assert all([a == b for a, b in zip(actual, expected)])