Merge pull request #116 from kurusugawa-computer/feature/labor

労務管理の作業時間を集計するツールを組み込む

annofabcli/__main__.py

@@ -4,6 +4,7 @@
 
 import annofabcli.annotation.subcommand_annotation
 import annofabcli.annotation_specs.subcommand_annotation_specs
+import annofabcli.experimental.subcommand_experimental
 import annofabcli.filesystem.subcommand_filesystem
 import annofabcli.input_data.subcommand_input_data
 import annofabcli.inspection_comment.subcommand_inspection_comment
@@ -51,6 +52,7 @@ def main(arguments: Optional[Sequence[str]] = None):
     annofabcli.supplementary.subcommand_supplementary.add_parser(subparsers)
 
     annofabcli.filesystem.subcommand_filesystem.add_parser(subparsers)
+    annofabcli.experimental.subcommand_experimental.add_parser(subparsers)
 
     if arguments is None:
         args = parser.parse_args()

annofabcli/common/cli.py

@@ -295,14 +295,14 @@ def add_csv_format(self, help_message: Optional[str] = None):
 
         self.parser.add_argument('--csv_format', type=str, help=help_message)
 
-    def add_output(self, help_message: Optional[str] = None):
+    def add_output(self, required: bool = False, help_message: Optional[str] = None):
         """
-        '--csv_format` 引数を追加
+        '--output` 引数を追加
         """
         if help_message is None:
             help_message = '出力先のファイルパスを指定します。指定しない場合は、標準出力に出力されます。'
 
-        self.parser.add_argument('-o', '--output', type=str, help=help_message)
+        self.parser.add_argument('-o', '--output', type=str, required=required, help=help_message)
 
     def add_query(self, help_message: Optional[str] = None):
         """

annofabcli/common/utils.py

@@ -211,7 +211,7 @@ def get_file_scheme_path(str_value: str) -> Optional[str]:
         return None
 
 
-def isoduration_to_hour(duration):
+def isoduration_to_hour(duration) -> float:
     """
     ISO 8601 duration を 時間に変換する
     Args:
@@ -224,6 +224,18 @@ def isoduration_to_hour(duration):
     return isodate.parse_duration(duration).total_seconds() / 3600
 
 
+def isoduration_to_minute(duration) -> float:
+    """
+    ISO 8601 duration を 分に変換する
+
+    Args:
+        duration (str): ISO 8601 Durationの文字
+    Returns:
+        変換後の分
+    """
+    return isodate.parse_duration(duration).total_seconds() / 60
+
+
 def allow_404_error(function):
     """
     Not Found Error(404)を無視(許容)して、処理する。Not Foundのとき戻りはNoneになる。

annofabcli/experimental/list_labor_worktime.py

@@ -0,0 +1,194 @@
+# flake8: noqa
+#  type: ignore
+# pylint: skip-file
+import argparse
+import datetime
+import logging
+from typing import Any, Dict, List, Tuple  # pylint: disable=unused-import
+
+import annofabapi
+from annofabapi.models import ProjectMemberRole
+
+import annofabcli
+import annofabcli.common.cli
+from annofabcli import AnnofabApiFacade
+from annofabcli.common.cli import AbstractCommandLineInterface, ArgumentParser, build_annofabapi_resource_and_login
+from annofabcli.experimental.utils import date_range, print_time_list_from_work_time_list
+from annofabcli.statistics.table import Table
+
+logger = logging.getLogger(__name__)
+
+
+class Database():
+    def __init__(self, annofab_service: annofabapi.Resource, project_id: str, organization_id: str):
+        self.annofab_service = annofab_service
+        self.project_id = project_id
+        self.organization_id = organization_id
+
+    def get_labor_control(self) -> List[Dict[str, Any]]:
+        labor_control_df = self.annofab_service.api.get_labor_control({
+            "organization_id": self.organization_id,
+            "project_id": self.project_id
+        })[0]
+        return labor_control_df
+
+    def get_account_statistics(self) -> List[Dict[str, Any]]:
+        account_statistics = self.annofab_service.api.get_account_statistics(self.project_id)[0]
+        return account_statistics
+
+    def get_project_members(self) -> dict:
+        project_members = self.annofab_service.api.get_project_members(project_id=self.project_id)[0]
+        return project_members["list"]
+
+
+class FutureTable(Table):
+    def __init__(self, database: Database, task_query_param=None):
+        super().__init__(database, task_query_param)
+        self.database = database
+
+    def create_labor_control_df(self):
+
+        labor_control = self.database.get_labor_control()
+        labor_control_list = []
+        for l in labor_control:
+            if l["values"] is None or l["values"]["working_time_by_user"] is None:
+                aw_plans = 0.0
+                aw_results = 0.0
+            else:
+                if l["values"]["working_time_by_user"]["plans"] is None:
+                    aw_plans = 0.0
+                else:
+                    aw_plans = int(l["values"]["working_time_by_user"]["plans"]) / 60000
+                if l["values"]["working_time_by_user"]["results"] is None:
+                    aw_results = 0.0
+                else:
+                    aw_results = int(l["values"]["working_time_by_user"]["results"]) / 60000
+
+            labor_control_list.append({
+                "account_id": l["account_id"],
+                "date": l["date"],
+                "aw_plans": aw_plans,
+                "aw_results": aw_results,
+                "username": self._get_username(l["account_id"]),
+                "af_time": 0.0
+            })
+
+        return labor_control_list
+
+    def create_account_statistics_df(self):
+        """
+        メンバごと、日ごとの作業時間
+        """
+        account_statistics = self.database.get_account_statistics()
+        all_histories = []
+        for account_info in account_statistics:
+
+            account_id = account_info['account_id']
+            histories = account_info['histories']
+
+            for history in histories:
+                history['af_time'] = annofabcli.utils.isoduration_to_minute(history['worktime'])
+                username = self._get_username(account_id)
+                history['account_id'] = account_id
+                history['username'] = username
+                history["aw_plans"] = 0.0
+                history["aw_results"] = 0.0
+
+            all_histories.extend(histories)
+        return all_histories
+
+    def create_afaw_time_df(self) -> Tuple[List[Any], List[Any]]:
+
+        account_statistics_df = self.create_account_statistics_df()
+        labor_control_df = self.create_labor_control_df()
+        username_list = []
+
+        for labor_control in labor_control_df:
+            for account_statistics in account_statistics_df:
+                if account_statistics["account_id"] == labor_control["account_id"] and account_statistics["date"] == \
+                        labor_control["date"]:
+                    labor_control["af_time"] = account_statistics["af_time"]
+                    labor_control["username"] = account_statistics["username"]
+                    if not account_statistics["username"] in username_list:
+                        username_list.append(account_statistics["username"])
+
+        return labor_control_df, username_list
+
+
+def get_organization_id_from_project_id(annofab_service: annofabapi.Resource, project_id: str) -> str:
+    """
+    project_idからorganization_idを返す
+    """
+    organization, _ = annofab_service.api.get_organization_of_project(project_id)
+    return organization["organization_id"]
+
+
+class ListLaborWorktime(AbstractCommandLineInterface):
+    """
+    労務管理画面の作業時間を出力する
+    """
+    def list_labor_worktime(self, project_id: str):
+        """
+        """
+
+        super().validate_project(project_id, project_member_roles=[ProjectMemberRole.OWNER])
+        # プロジェクト or 組織に対して、必要な権限が付与されているかを確認
+
+        organization_id = get_organization_id_from_project_id(self.service, project_id)
+        database = Database(self.service, project_id, organization_id)
+        # Annofabから取得した情報に関するデータベースを取得するクラス
+        table_obj = FutureTable(database=database)
+        # Databaseから取得した情報を元にPandas DataFrameを生成するクラス
+        #     チェックポイントファイルがあること前提
+        return table_obj.create_afaw_time_df()
+
+    def main(self):
+        args = self.args
+        start_date = datetime.datetime.strptime(args.start_date, '%Y-%m-%d').date()
+        end_date = datetime.datetime.strptime(args.end_date, '%Y-%m-%d').date()
+        date_list = date_range(start_date, end_date)
+
+        afaw_time_list = []
+        project_members_list = []
+        for i, project_id in enumerate(args.project_id):
+            logger.debug(f"{i + 1} 件目: project_id = {project_id}")
+            afaw_time_df, project_members = self.list_labor_worktime(project_id)
+            afaw_time_list.append(afaw_time_df)
+            project_members_list.extend(project_members)
+        print_time_list = print_time_list_from_work_time_list(list(set(project_members_list)), afaw_time_list,
+                                                              date_list)
+
+        output_lines: List[str] = []
+        output_lines.append(f"Start: , {start_date},  End: , {end_date}")
+        output_lines.extend([",".join([str(cell) for cell in row]) for row in print_time_list])
+        annofabcli.utils.output_string("\n".join(output_lines), args.output)
+
+
+def main(args):
+    service = build_annofabapi_resource_and_login()
+    facade = AnnofabApiFacade(service)
+    ListLaborWorktime(service, facade, args).main()
+
+
+def parse_args(parser: argparse.ArgumentParser):
+    argument_parser = ArgumentParser(parser)
+
+    parser.add_argument(
+        '-p', '--project_id', type=str, required=True, nargs='+',
+        help="集計対象のプロジェクトのproject_idを指定します。複数指定した場合は合計値を出力します。"
+        "`file://`を先頭に付けると、project_idの一覧が記載されたファイルを指定できます。")
+    parser.add_argument("--start_date", type=str, required=True, help="集計開始日(%%Y-%%m-%%d)")
+    parser.add_argument("--end_date", type=str, required=True, help="集計終了日(%%Y-%%m-%%d)")
+
+    argument_parser.add_output(required=True)
+
+    parser.set_defaults(subcommand_func=main)
+
+
+def add_parser(subparsers: argparse._SubParsersAction):
+    subcommand_name = "list_labor_worktime"
+    subcommand_help = "労務管理画面の作業時間を出力します。"
+    description = ("作業者ごとに、「作業者が入力した実績時間」と「AnnoFabが集計した作業時間」の差分を出力します。")
+
+    parser = annofabcli.common.cli.add_parser(subparsers, subcommand_name, subcommand_help, description)
+    parse_args(parser)

annofabcli/experimental/subcommand_experimental.py

@@ -0,0 +1,22 @@
+import argparse
+
+import annofabcli
+import annofabcli.common.cli
+import annofabcli.experimental.list_labor_worktime
+
+
+def parse_args(parser: argparse.ArgumentParser):
+
+    subparsers = parser.add_subparsers(dest='subcommand_name')
+
+    # サブコマンドの定義
+    annofabcli.experimental.list_labor_worktime.add_parser(subparsers)  # type: ignore
+
+
+def add_parser(subparsers: argparse._SubParsersAction):
+    subcommand_name = "experimental"
+    subcommand_help = "アルファ版のサブコマンド"
+    description = "アルファ版のサブコマンド。予告なしに削除されたり、コマンドライン引数が変わったりします。"
+
+    parser = annofabcli.common.cli.add_parser(subparsers, subcommand_name, subcommand_help, description)
+    parse_args(parser)

annofabcli/experimental/utils.py

@@ -0,0 +1,87 @@
+# flake8: noqa
+#  type: ignore
+# pylint: skip-file
+from datetime import date, timedelta
+from typing import Any, Callable, Dict, List, Optional, Tuple, Union  # pylint: disable=unused-import
+
+
+def date_range(start_date: date, end_date: date):
+    diff = (end_date - start_date).days + 1
+    return (start_date + timedelta(i) for i in range(diff))
+
+
+def print_time_list_from_work_time_list(username_list: list, work_time_lists: list, date_list: list):
+    print_time_list = []
+    new_header_name_list = []
+    new_header_item_list = []
+    new_footer_item_list = []
+    new_header_name_list.append("")
+    new_header_item_list.append("date / time")
+    new_footer_item_list.append("total_time")
+    username_footer_list: Dict[str, Any] = {}
+
+    for username in username_list:
+        new_header_name_list.append(username)
+        new_header_name_list.append("")
+        new_header_name_list.append("")
+        new_header_name_list.append("")
+        new_header_name_list.append("")
+        new_header_item_list.append("aw_plans")
+        new_header_item_list.append("aw_results")
+        new_header_item_list.append("af_time")
+        new_header_item_list.append("diff")
+        new_header_item_list.append("diff_per")
+        username_footer_list[username] = {}
+        username_footer_list[username]["aw_plans"] = 0.0
+        username_footer_list[username]["aw_results"] = 0.0
+        username_footer_list[username]["af_time"] = 0.0
+
+    print_time_list.append(new_header_name_list)
+    print_time_list.append(new_header_item_list)
+
+    for day in date_list:
+        new_line = []
+        new_line.append(day.strftime("%Y-%m-%d"))
+        for username in username_list:
+            aw_plans = 0.0
+            aw_results = 0.0
+            af_time = 0.0
+
+            for work_time_list in work_time_lists:
+                for work_time in work_time_list:
+                    if username == work_time["username"] and day.strftime("%Y-%m-%d") == work_time["date"]:
+                        aw_plans += work_time["aw_plans"]
+                        aw_results += work_time["aw_results"]
+                        af_time += work_time["af_time"]
+
+            new_line.append(round(aw_plans, 2))
+            new_line.append(round(aw_results, 2))
+            new_line.append(round(af_time, 2))
+            diff = round(aw_results - af_time, 2)
+            diff_per = round(diff / aw_results, 2) if aw_results != 0.0 else 0.0
+            new_line.append(diff)
+            new_line.append(diff_per)
+            username_footer_list[username]["aw_plans"] += aw_plans
+            username_footer_list[username]["aw_results"] += aw_results
+            username_footer_list[username]["af_time"] += af_time
+
+        print_time_list.append(new_line)
+
+    for username in username_list:
+        new_footer_item_list.append(round(username_footer_list[username]["aw_plans"], 2))
+        new_footer_item_list.append(round(username_footer_list[username]["aw_results"], 2))
+        new_footer_item_list.append(round(username_footer_list[username]["af_time"], 2))
+        total_diff = round(username_footer_list[username]["aw_results"] - username_footer_list[username]["af_time"], 2)
+        total_diff_per = round(total_diff / username_footer_list[username]["aw_results"], 2) \
+            if (username_footer_list[username]["aw_results"] != 0.0) else 0.0
+        new_footer_item_list.append(total_diff)
+        new_footer_item_list.append(total_diff_per)
+
+    print_time_list.append(new_footer_item_list)
+
+    return print_time_list
+
+
+def print_time_list_csv(print_time_list: list) -> None:
+    for time_list in print_time_list:
+        print(*time_list, sep=",")