scriptsに、CSVから`task update_metadata --metadata_by_task_id`に渡せるJSONファイルを生成するスクリプトを追加

サンプルコード

ほぼこれが利用できるはず。

import argparse
import json
from argparse import ArgumentParser
from pathlib import Path

import pandas
from loguru import logger

from src.common.cli import PrettyHelpFormatter, create_parent_parser
from src.common.utils import configure_loguru, log_exception


def create_metadata_dict_by_task_id(input_csv: Path, task_json: Path, *, target_task_id_prefix: str | None = None) -> dict[str, dict[str, int]]:
    """
    優先順位が記載されたCSVと、タスク一覧のCSVから、以下のdictを生成します。
    * key: タスクID
    * value: 設定するメタデータ
    """
    df_input = pandas.read_csv(input_csv)
    task_list = json.loads(task_json.read_text())

    result = {}
    for prefix, priority in zip(df_input["task_id_prefix"], df_input["priority"], strict=True):
        if target_task_id_prefix and not prefix.startswith(target_task_id_prefix):
            continue

        sub_result = {}
        for task in task_list:
            task_id = task["task_id"]
            if task_id.startswith(prefix):
                sub_result[task_id] = {"priority": priority}

        logger.info(f"'{prefix}'に部分一致するタスクが{len(sub_result)}件見つかりました。 :: priority='{priority}'")
        result.update(sub_result)

    return result


def create_parser() -> argparse.ArgumentParser:
    parser = ArgumentParser(
        description="タスクの優先順位が記載されたCSVファイルから、`annofabcli task update_metadata`コマンドの`--metadata_by_task_id`に渡せるJSONを出力します。",
        formatter_class=PrettyHelpFormatter,
        parents=[create_parent_parser()],
    )

    parser.add_argument("--priority_csv", type=Path, required=True, help="タスクの優先順位が記載されているCSVファイル")
    parser.add_argument("--task_json", type=Path, required=True, help="タスクの一覧が記載されているJSONファイル（`annofabcli task download`の出力ファイル）")
    parser.add_argument("-o", "--output_file", type=Path, required=True, help="出力先ファイルのパス")

    parser.add_argument(
        "--task_id_prefix",
        type=str,
        help="`--priority_csv`に指定したCSVの中で、前方一致に該当したtask_id_prefixが対象になります。指定しない場合、`--priority_csv`に記載されている全てが対象になります。",
    )

    return parser


@log_exception(logger=logger)
def main() -> None:
    args = create_parser().parse_args()
    configure_loguru(is_verbose=args.verbose)

    metadata_dict = create_metadata_dict_by_task_id(args.priority_csv, args.task_json, target_task_id_prefix=args.task_id_prefix)
    output_file: Path = args.output_file
    logger.info(f"{len(metadata_dict)} 件のタスクに対してメタデータ'priority'が記載されたJSONを'{output_file}'に出力します。")

    output_file.parent.mkdir(exist_ok=True, parents=True)
    output_file.write_text(json.dumps(metadata_dict, ensure_ascii=False))


if __name__ == "__main__":
    main()

なぜコマンドでなくスクリプトにしたのか

task_idのprefixで前方一致したいかどうかは、プロジェクトの運用によって変わりそうだから。
常にサポートし続けるのは、少し厳しい内容だった。
またプログラミングの扱いに慣れている人ならば、自分で書いた方が早いケースもありそうだったので。

とはいえ、よく利用する運用だったので、スクリプトとしては残しておきたかった。

Start by checking the repository's scripts/ directory and existing Python script conventions, then adapt the sample code using the shared CLI helpers in src.common. Run it with a priority CSV and task JSON from task download, and verify that the output JSON can be passed to task update_metadata with --metadata_by_task_id.