第161回 AIへ渡すCSVを必要列だけに絞る:Pythonで作る許可列エクスポートゲート

問い合わせCSVをAIで分類するとき、氏名やメールアドレスまで入力に残す必要がない場合があります。手作業で列を削除する運用では、削除漏れや担当者ごとの手順差を防げません。

この記事では、ポリシーで許可した列だけを別のCSVへ書き出すゲートを、Python標準ライブラリだけで作ります。このスクリプトはAIサービスへ通信しません。生成したCSVを外部へ渡してよいかは、組織の規則、契約、データの内容に基づいて別途確認します。

今回作る処理の契約

  • 出力列は allowed_columns に記載した列だけとし、記載順も維持する
  • 禁止列が許可リストにも含まれていたら、ポリシーの誤りとして拒否する
  • ヘッダーの重複、許可列の不足、行の列数不一致を拒否する
  • 既存の出力ファイルを上書きしない
  • ヘルプ表示を除き、標準出力には1行のJSONを出す
  • データ行のセル値は結果JSONへ出さない

forbidden_columns は、ポリシー編集時の事故を防ぐための二重チェックです。入力CSVに禁止列が存在すること自体はエラーではありません。その列を出力対象へ含めようとした場合に停止します。

終了コード 意味
0 出力完了
2 引数、入力CSV、またはポリシーを拒否
3 ファイル操作または予期しない実行エラー
130 Ctrl+Cによる中断

1. サンプルCSVを用意する

動作確認用の架空データを source.csv として保存します。文字コードはUTF-8です。

case_id,customer_name,email,category,request_text,internal_note
C-001,山田花子,hanako@example.com,配送,到着日を確認したい,返金履歴あり
C-002,佐藤太郎,taro@example.com,請求,領収書を再発行したい,本人確認済み

2. 許可列と禁止列を定義する

次の内容を export-policy.json として保存します。出力CSVの列順は allowed_columns の順番になります。

{
  "allowed_columns": [
    "case_id",
    "category",
    "request_text"
  ],
  "forbidden_columns": [
    "customer_name",
    "email",
    "phone",
    "address",
    "internal_note"
  ]
}

ここでの列名は例です。実際には、自社のCSV仕様と利用目的に合わせて明示的に決めます。自由記述欄には氏名や連絡先が入力される可能性があるため、列名だけの検査では安全性を保証できません。

3. エクスポートゲートを実装する

次のコードを export_allowed_csv.py として保存します。Python 3.10以降を想定し、外部ライブラリは使いません。

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import csv
import json
import os
import sys
from pathlib import Path
from typing import Any

EXIT_OK = 0
EXIT_REJECTED = 2
EXIT_OPERATIONAL_ERROR = 3
EXIT_INTERRUPTED = 130


class RejectedError(Exception):
    """引数、入力、またはポリシーを拒否した。"""


class OperationalError(Exception):
    """ファイル操作など、実行環境のエラー。"""


class JsonArgumentParser(argparse.ArgumentParser):
    def error(self, message: str) -> None:
        raise RejectedError(f"引数エラー: {message}")


def emit(payload: dict[str, Any]) -> None:
    print(json.dumps(payload, ensure_ascii=False, separators=(",", ":")))


def find_duplicates(names: list[str]) -> list[str]:
    seen: set[str] = set()
    duplicates: set[str] = set()
    for name in names:
        if name in seen:
            duplicates.add(name)
        seen.add(name)
    return sorted(duplicates)


def unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise RejectedError(
                f"ポリシーJSONに重複キーがあります: {key}"
            )
        result[key] = value
    return result


def read_name_list(
    data: dict[str, Any], key: str, *, allow_empty: bool
) -> list[str]:
    value = data[key]
    if not isinstance(value, list):
        raise RejectedError(f"{key} は配列で指定してください")
    if not allow_empty and not value:
        raise RejectedError(f"{key} を1件以上指定してください")

    names: list[str] = []
    for item in value:
        if (
            not isinstance(item, str)
            or not item
            or item != item.strip()
            or "\r" in item
            or "\n" in item
        ):
            raise RejectedError(
                f"{key} の列名に空欄、前後空白、または改行があります"
            )
        names.append(item)

    duplicates = find_duplicates(names)
    if duplicates:
        raise RejectedError(
            f"{key} に重複があります: {', '.join(duplicates)}"
        )
    return names


def load_policy(path: Path) -> list[str]:
    try:
        text = path.read_text(encoding="utf-8-sig")
    except UnicodeDecodeError as exc:
        raise RejectedError("ポリシーはUTF-8で読めません") from exc
    except OSError as exc:
        raise OperationalError(f"ポリシーを読めません: {path}") from exc

    try:
        raw = json.loads(text, object_pairs_hook=unique_object)
    except json.JSONDecodeError as exc:
        raise RejectedError(
            f"ポリシーJSONの構文エラーです(行{exc.lineno}、列{exc.colno})"
        ) from exc

    if not isinstance(raw, dict):
        raise RejectedError("ポリシーの最上位はオブジェクトにしてください")

    expected_keys = {"allowed_columns", "forbidden_columns"}
    actual_keys = set(raw)
    if actual_keys != expected_keys:
        details: list[str] = []
        missing = sorted(expected_keys - actual_keys)
        extra = sorted(actual_keys - expected_keys)
        if missing:
            details.append(f"不足: {', '.join(missing)}")
        if extra:
            details.append(f"未定義: {', '.join(extra)}")
        raise RejectedError(
            f"ポリシーのキーが不正です({'; '.join(details)})"
        )

    allowed = read_name_list(raw, "allowed_columns", allow_empty=False)
    forbidden = set(
        read_name_list(raw, "forbidden_columns", allow_empty=True)
    )

    overlap = sorted(set(allowed) & forbidden)
    if overlap:
        raise RejectedError(
            "allowed_columns と forbidden_columns が重複しています: "
            + ", ".join(overlap)
        )
    return allowed


def validate_header(
    header: list[str] | None, allowed: list[str]
) -> list[str]:
    if header is None:
        raise RejectedError("入力CSVが空です")

    invalid = [
        name
        for name in header
        if not name
        or name != name.strip()
        or "\r" in name
        or "\n" in name
    ]
    if invalid:
        raise RejectedError(
            "CSVヘッダーに空欄、前後空白、または改行があります"
        )

    duplicates = find_duplicates(header)
    if duplicates:
        raise RejectedError(
            f"CSVヘッダーに重複があります: {', '.join(duplicates)}"
        )

    missing = [name for name in allowed if name not in header]
    if missing:
        raise RejectedError(
            f"入力CSVに許可列がありません: {', '.join(missing)}"
        )
    return header


def same_path(left: Path, right: Path) -> bool:
    try:
        return left.resolve(strict=False) == right.resolve(strict=False)
    except (OSError, RuntimeError) as exc:
        raise OperationalError("入出力パスを解決できません") from exc


def cleanup_partial(path: Path) -> None:
    try:
        path.unlink()
    except FileNotFoundError:
        return
    except OSError as exc:
        raise OperationalError(
            f"不完全な出力ファイルを削除できません: {path}"
        ) from exc


def export_allowed_csv(
    source_path: Path, policy_path: Path, output_path: Path
) -> dict[str, Any]:
    if same_path(source_path, output_path):
        raise RejectedError("入力元と出力先に同じパスは指定できません")
    if output_path.exists() or output_path.is_symlink():
        raise RejectedError(f"出力先が既に存在します: {output_path}")
    if not output_path.parent.exists() or not output_path.parent.is_dir():
        raise OperationalError(
            f"出力先ディレクトリを利用できません: {output_path.parent}"
        )

    allowed = load_policy(policy_path)
    output_created = False
    completed = False
    reader = None
    header: list[str] = []
    row_count = 0

    try:
        with source_path.open(
            "r", encoding="utf-8-sig", newline=""
        ) as source:
            reader = csv.DictReader(source, strict=True)
            header = validate_header(reader.fieldnames, allowed)

            with output_path.open(
                "x", encoding="utf-8", newline=""
            ) as target:
                output_created = True
                writer = csv.DictWriter(
                    target,
                    fieldnames=allowed,
                    extrasaction="raise",
                    lineterminator="\n",
                )
                writer.writeheader()

                for row in reader:
                    if None in row:
                        raise RejectedError(
                            "列数がヘッダーより多い行があります: "
                            f"物理行{reader.line_num}"
                        )

                    missing_values = [
                        name for name in header if row.get(name) is None
                    ]
                    if missing_values:
                        raise RejectedError(
                            "列数がヘッダーより少ない行があります: "
                            f"物理行{reader.line_num}"
                        )

                    writer.writerow(
                        {name: row[name] for name in allowed}
                    )
                    row_count += 1

                target.flush()
                os.fsync(target.fileno())

        completed = True

    except FileExistsError as exc:
        raise RejectedError(
            f"出力先が既に存在します: {output_path}"
        ) from exc
    except UnicodeDecodeError as exc:
        raise RejectedError(
            "入力CSVはUTF-8(BOM可)として読めません"
        ) from exc
    except csv.Error as exc:
        line_number = reader.line_num if reader is not None else 0
        raise RejectedError(
            f"CSV構文エラーです(物理行{line_number})"
        ) from exc
    except OSError as exc:
        target_name = exc.filename or "対象不明"
        raise OperationalError(
            f"ファイル操作に失敗しました: {target_name}"
        ) from exc
    finally:
        if output_created and not completed:
            cleanup_partial(output_path)

    return {
        "rows": row_count,
        "input_columns": len(header),
        "output_columns": allowed,
        "dropped_columns": len(header) - len(allowed),
        "output": str(output_path),
    }


def build_parser() -> argparse.ArgumentParser:
    parser = JsonArgumentParser(
        description="許可した列だけをCSVへ書き出します"
    )
    parser.add_argument("source", type=Path, help="入力CSV")
    parser.add_argument("policy", type=Path, help="ポリシーJSON")
    parser.add_argument("output", type=Path, help="新規出力CSV")
    return parser


def main(argv: list[str] | None = None) -> int:
    try:
        args = build_parser().parse_args(argv)
        report = export_allowed_csv(
            args.source, args.policy, args.output
        )
    except RejectedError as exc:
        emit({"status": "rejected", "reason": str(exc)})
        return EXIT_REJECTED
    except OperationalError as exc:
        emit({"status": "error", "reason": str(exc)})
        return EXIT_OPERATIONAL_ERROR
    except KeyboardInterrupt:
        emit({"status": "error", "reason": "処理が中断されました"})
        return EXIT_INTERRUPTED
    except Exception as exc:
        emit(
            {
                "status": "error",
                "reason": f"予期しないエラー: {type(exc).__name__}",
            }
        )
        return EXIT_OPERATIONAL_ERROR

    emit({"status": "ok", **report})
    return EXIT_OK


if __name__ == "__main__":
    sys.exit(main())

4. 実行結果を確認する

出力先には、まだ存在しないファイル名を指定します。

python3 export_allowed_csv.py source.csv export-policy.json ai_input.csv
rc=$?
echo "exit=$rc"

スクリプトが標準出力へ出すJSONは次の1行です。

{"status":"ok","rows":2,"input_columns":6,"output_columns":["case_id","category","request_text"],"dropped_columns":3,"output":"ai_input.csv"}

続いて、シェルの echo が exit=0 を表示します。生成された ai_input.csv は次の内容です。

case_id,category,request_text
C-001,配送,到着日を確認したい
C-002,請求,領収書を再発行したい

元の行順は維持されます。空文字のセルは許可しますが、列自体が不足している行は拒否します。データ行が0件でも、ヘッダーが正しければ正常終了します。

5. ポリシーの誤りを止められるか確認する

email を許可列と禁止列の両方へ入れた bad-policy.json を用意します。

{
  "allowed_columns": [
    "case_id",
    "email",
    "category",
    "request_text"
  ],
  "forbidden_columns": [
    "customer_name",
    "email",
    "phone",
    "address",
    "internal_note"
  ]
}

既存ファイルと衝突しない出力名で実行します。

python3 export_allowed_csv.py source.csv bad-policy.json bad.csv
rc=$?
echo "exit=$rc"
test ! -e bad.csv
echo "output_absent=$?"

標準出力と終了コードは次のようになります。output_absent=0 は、bad.csv が存在しないことを確認できたという意味です。

{"status":"rejected","reason":"allowed_columns と forbidden_columns が重複しています: email"}
exit=2
output_absent=0

運用へ組み込む順序

  1. 処理目的に必要な列を担当者とデータ管理者が決める
  2. 許可列と禁止列をポリシーJSONでレビューする
  3. 元CSVから別名のCSVを生成する
  4. 終了コードが0であることと、結果JSONの件数・列名を確認する
  5. 生成ファイルの内容を確認してから、承認済みの処理へ渡す
  6. 生成ファイルは組織の保管・削除ルールに従って扱う

後続処理は、ファイルの存在だけではなく終了コード0を開始条件にします。処理中も出力ファイルは見えるため、同じ出力先を監視する別処理がある場合は、完了前に読み込ませない設計が必要です。

この実装で保証しないこと

  • 許可した自由記述欄に含まれる氏名、電話番号、秘密情報の検出
  • case_idなど、別データと照合できる識別子の匿名化
  • AIサービスへの送信可否や法令・契約への適合判断
  • 表計算ソフトで開く場合の数式注入対策
  • タブ区切りやセミコロン区切りなど、カンマ区切り以外の形式
  • 同じ出力先へ複数プロセスが同時に書き込む運用
  • 強制終了や電源断のあとに不完全なファイルが残らないこと

通常の検証エラーやCtrl+Cでは、作成途中と認識できた出力を削除します。ただし、プロセスの強制終了や電源断では不完全なファイルが残る可能性があります。再実行前に出力ファイルを確認し、不要と判断できた場合だけ明示的に削除してください。

まとめ

AIへ渡すデータを減らすときは、「削除する列」を毎回選ぶより、「出してよい列」を固定する方が手順を再現しやすくなります。このゲートは完全な匿名化ツールではありません。それでも、許可列、ポリシーの矛盾、CSV構造、上書き防止、終了コードを検査し、列の削除漏れを機械的に止められます。

第160回 CSVを表計算ソフトへ渡す前に止める:Pythonで作る数式注入チェック

第159回では、CSVをUTF-8へ正規化する入口を作りました。しかし、文字コードが正しくても、表計算ソフトへ渡す前に確認したいことがあります。外部から受け取った文字列が特定の記号で始まると、文字列ではなく数式として解釈される可能性があるためです。

Pythonのcsvモジュールが数式を実行するわけではありません。問題になるのは、担当者や取引先がCSVを表計算ソフトで開く段階です。今回は、外部入力が入るテキスト列を検査し、数式候補があれば終了コード3で配布を止めるCLIを作ります。入力ファイルは変更しません。

CSVの引用符だけでは防げない

CSVの引用符は、カンマや改行を含む値を一つのセルとして扱うための構文です。たとえばCSV内で"=1+1"と書かれていても、読み込み後の値は=1+1です。引用符で囲むことと、数式としての解釈を止めることは別の処理です。

このスクリプトでは、先頭の半角スペースを除いた最初の文字が、次のいずれかなら数式候補として検出します。

  • 半角の=、+、-、@
  • タブ、CR、LF
  • 全角の=、+、-、@

全角文字は、日本語環境を考慮した保守的な検査です。実際の解釈は、表計算ソフト、バージョン、地域設定、読み込み方法によって変わります。この検査だけで、すべての環境における安全性を保証することはできません。

入出力の決まり

  • 入力はヘッダー付き、カンマ区切りのUTF-8 CSVとする。UTF-8 BOMの有無はどちらも受け付ける。
  • --text-columnで指定した列だけを検査する。
  • 外部入力が入るテキスト列は漏れなく指定する。
  • 数値列は事前のスキーマ検査で数値だけに制限する。負数を許可するためだけに、未検証の列を検査対象から外さない。
  • セルの元の値はJSONへ出さず、論理レコード番号、列名、検出文字だけを記録する。
  • 入力ファイルは変更せず、変換ファイルも作らない。

数式候補を自動的に書き換えないのは、先頭へのアポストロフィやタブの追加が、すべての表計算ソフトと後続処理で同じ意味になるとは限らないためです。検出時は配布を止め、元データまたはCSVを作る処理を修正します。

検査スクリプト

次のコードをcsv_formula_guard.pyとして保存します。Python 3.8以降の標準ライブラリだけを使います。

#!/usr/bin/env python3
import argparse
import csv
import json
import sys
from pathlib import Path


DANGEROUS_PREFIXES = frozenset("=+-@\t\r\n=+-@")
LEADING_SPACES = " "
MAX_DETAILS = 100


class InputError(Exception):
    pass


def formula_trigger(value):
    candidate = value.lstrip(LEADING_SPACES)
    if not candidate:
        return None

    first = candidate[0]
    if first in DANGEROUS_PREFIXES:
        return first
    return None


def validate_header(header):
    if not header:
        raise InputError("ヘッダーがありません")
    if any(not name.strip() for name in header):
        raise InputError("空の列名があります")
    if any(name != name.strip() for name in header):
        raise InputError("列名の前後に空白があります")
    if len(header) != len(set(header)):
        raise InputError("列名が重複しています")


def unique_columns(names):
    result = []
    for name in names:
        if name not in result:
            result.append(name)
    return result


def select_columns(header, names):
    missing = [name for name in names if name not in header]
    if missing:
        raise InputError("対象列がありません: " + ", ".join(missing))
    return [(header.index(name), name) for name in names]


def scan_csv(path, columns):
    finding_count = 0
    details = []

    try:
        with path.open("r", encoding="utf-8-sig", newline="") as source:
            reader = csv.reader(source, strict=True)
            try:
                header = next(reader)
            except StopIteration as exc:
                raise InputError("CSVが空です") from exc

            validate_header(header)
            selected = select_columns(header, columns)

            for record_number, row in enumerate(reader, start=2):
                if len(row) != len(header):
                    raise InputError(
                        f"レコード{record_number}: 列数が{len(row)}です"
                        f"(期待値: {len(header)})"
                    )

                for index, column_name in selected:
                    trigger = formula_trigger(row[index])
                    if trigger is None:
                        continue

                    finding_count += 1
                    if len(details) < MAX_DETAILS:
                        details.append({
                            "record": record_number,
                            "column": column_name,
                            "trigger": trigger,
                        })

    except UnicodeDecodeError as exc:
        raise InputError(f"{path}: UTF-8として読めません") from exc
    except csv.Error as exc:
        raise InputError(f"{path}: CSV構文エラー: {exc}") from exc

    return finding_count, details


def build_parser():
    parser = argparse.ArgumentParser(
        description="CSVの指定列にある数式候補を検査します"
    )
    parser.add_argument("input", type=Path)
    parser.add_argument(
        "--text-column",
        dest="text_columns",
        action="append",
        required=True,
        help="検査する列名。複数回指定できます",
    )
    return parser


def print_report(report):
    print(json.dumps(
        report,
        ensure_ascii=False,
        separators=(",", ":"),
    ))


def main():
    args = build_parser().parse_args()
    columns = unique_columns(args.text_columns)

    try:
        finding_count, details = scan_csv(args.input, columns)
    except (InputError, OSError) as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 1

    report = {
        "status": "blocked" if finding_count else "ok",
        "input": str(args.input),
        "checked_columns": columns,
        "finding_count": finding_count,
        "findings": details,
        "findings_truncated": finding_count > len(details),
    }
    print_report(report)
    return 3 if finding_count else 0


if __name__ == "__main__":
    sys.exit(main())

全レコードをリストへ保存せず、1レコードずつ検査します。詳細は最大100件に制限しますが、finding_countは最後まで数えます。

検出結果を再現する

次の内容をUTF-8のsample.csvとして保存します。amount列は、別工程で数値として検証済みという前提です。

id,customer_name,memo,amount
1,株式会社A,折り返し連絡,1200
2,=1+1,通常,900
3,株式会社C," +SUM(1,2)",-500

外部入力が入るcustomer_name列とmemo列を検査します。

python3 csv_formula_guard.py sample.csv --text-column customer_name --text-column memo

標準出力は次の1行です。

{"status":"blocked","input":"sample.csv","checked_columns":["customer_name","memo"],"finding_count":2,"findings":[{"record":3,"column":"customer_name","trigger":"="},{"record":4,"column":"memo","trigger":"+"}],"findings_truncated":false}

終了コードは3です。recordは物理的な行番号ではなく、ヘッダーを1番としたCSV上の論理レコード番号です。セル内に改行があっても、一つのレコードとして数えます。

amount列の-500は検査対象ではないため、数式候補には数えません。ただし、数値列へ文字列を混入できる状態であれば、この運用は安全ではありません。先に整数や小数として解析できることを検査してください。

終了コードを後続処理へ渡す

終了コード 意味 後続処理
0 数式候補がない 次の工程へ進める
1 ファイル、文字コード、CSV構文、ヘッダーまたは列数のエラー 入力を修正して再実行する
2 コマンド引数の誤り ファイル名と列指定を確認する
3 数式候補を検出した 配布を止め、元データを確認する

入力エラーは標準エラー出力へ書き、JSONは出力しません。後続処理ではJSONの文言だけで判断せず、終了コードを確認します。

業務フローへ入れる順番

  1. 第159回の入口で文字コードをUTF-8へそろえる。
  2. 固定ヘッダー、列数、必須項目、数値型などをスキーマ検査する。
  3. 外部入力が入るテキスト列をすべて指定して、このスクリプトを実行する。
  4. 終了コード3なら配布を止め、該当レコードの元データを修正する。
  5. 修正後に同じ条件で再実行する。
  6. 終了コード0を確認したうえで、実際の受け渡し先と同じ方法でも読み込み結果を確認する。

この検査の制約

  • 対象は指定した列だけです。列指定の漏れは検出できません。
  • ヘッダー自体は数式候補として検査しません。外部入力から列名を作らず、固定スキーマと照合してください。
  • CSVの区切り文字はカンマ、引用符はダブルクォートというcsvモジュールの既定設定です。
  • strict=Trueでも、あらゆる不正形式を判定できるわけではありません。列数と固定スキーマの検査を併用します。
  • 検出文字の集合は保守的な規則です。表計算ソフト固有の挙動や将来の変更までは判定しません。
  • 終了コード0は、この規則に一致する値がなかったことを示すだけです。受け渡し先での安全性を証明するものではありません。

最初に行うのは、顧客名、備考、問い合わせ内容など、外部入力が入る列の洗い出しです。検出後に自動変換するのではなく、終了コード3で配布を止めるところまで業務手順に組み込みます。

確認資料

第159回 CSVの文字コードを推測しない:Pythonで作るUTF-8正規化ゲート

第158回では、CSVの列構成をv1からv2へ移行する処理を扱いました。しかし、その前段で文字コードを正しく扱えなければ、移行処理には進めません。

今回は、受け取ったCSVをバイト列で読み、まずUTF-8として厳密に検査します。UTF-8でなかった場合は、明示的に許可したときだけCP932として読みます。文字コードを自動推測せず、受け入れる形式を業務ルールとして限定する設計です。

今回の入出力契約

項目 仕様
既定の入力 BOMなしUTF-8、または先頭にUTF-8 BOMが1個だけあるUTF-8
任意の入力 --allow-cp932を指定した場合のみCP932
対象外 UTF-16、UTF-32、およびUTF-8 BOMが2個以上連続する入力
出力 BOMなしUTF-8
CSV条件 1行目に少なくとも1つ値のあるヘッダーがあり、すべてのレコードがヘッダーと同じ列数であること
上書き 既存ファイルへの上書きは禁止
変換成功時 標準出力へ1行のJSON、終了コード0
スクリプト内で検出したエラー 標準エラー出力へ1行のJSON、終了コード2~5

CP932は自動判定するのではなく、UTF-8でなかった場合の代替候補です。入力元がCP932を出力すると確認できている業務だけで有効にします。

UTF-8 BOMは先頭に1個だけ許可します。複数のBOMを成功扱いにすると、utf-8-sigで除去されなかったBOMが出力へ残るため、この実装では文字コードエラーとして拒否します。

処理の順序

  1. 入力を文字列ではなくバイト列として読む
  2. UTF-16またはUTF-32のBOMがあれば対象外として停止する
  3. UTF-8 BOMが2個以上連続していれば停止する
  4. 単一のUTF-8 BOM、BOMなしUTF-8、許可時のCP932の順で厳密にデコードする
  5. CSVとして解析し、ヘッダーと列数を確認する
  6. BOMなしUTF-8へ変換し、新しいファイルとして保存する
  7. 入力文字コード、列数、データ行数をJSONで報告する

CSVの検査にはcsv.readerを使いますが、出力をcsv.writerでは再生成しません。引用符や改行コードを組み直さず、デコードした文字列をUTF-8へ変換します。単一のBOMが付いたUTF-8入力では、そのBOMだけを除いて内容を維持します。

標準ライブラリだけで実装する

次のコードをnormalize_csv_encoding.pyとして保存します。型注釈にはPython 3.9以降の記法を使用しています。

import argparse
import codecs
import csv
import io
import json
import sys
from pathlib import Path

EXIT_OK = 0
EXIT_INPUT = 2
EXIT_ENCODING = 3
EXIT_CSV = 4
EXIT_OUTPUT = 5


class EncodingRejected(Exception):
    pass


class CsvRejected(Exception):
    pass


class OutputRejected(Exception):
    pass


def decode_input(data: bytes, allow_cp932: bool) -> tuple[str, str]:
    unsupported_boms = (
        codecs.BOM_UTF32_LE,
        codecs.BOM_UTF32_BE,
        codecs.BOM_UTF16_LE,
        codecs.BOM_UTF16_BE,
    )
    if data.startswith(unsupported_boms):
        raise EncodingRejected(
            'UTF-16/32は対象外です。送信元でUTF-8またはCP932として書き出してください'
        )

    if data.startswith(codecs.BOM_UTF8 * 2):
        raise EncodingRejected('UTF-8 BOMが複数連続しているため受け入れません')

    if data.startswith(codecs.BOM_UTF8):
        try:
            return data.decode('utf-8-sig', errors='strict'), 'utf-8-sig'
        except UnicodeDecodeError as exc:
            raise EncodingRejected(
                'UTF-8 BOMがありますが、本文をUTF-8としてデコードできません'
            ) from exc

    try:
        return data.decode('utf-8', errors='strict'), 'utf-8'
    except UnicodeDecodeError as utf8_error:
        if not allow_cp932:
            raise EncodingRejected(
                '入力はUTF-8ではありません。CP932を受け入れる場合だけ '
                '--allow-cp932 を指定してください'
            ) from utf8_error

    try:
        return data.decode('cp932', errors='strict'), 'cp932'
    except UnicodeDecodeError as cp932_error:
        raise EncodingRejected(
            '入力をUTF-8またはCP932として厳密にデコードできません'
        ) from cp932_error


def inspect_csv(text: str) -> tuple[int, int]:
    if '\x00' in text:
        raise CsvRejected('NUL文字を含むためCSVとして受け入れません')

    reader = csv.reader(io.StringIO(text, newline=''), strict=True)

    try:
        header = next(reader)
    except StopIteration as exc:
        raise CsvRejected('CSVが空です') from exc
    except csv.Error as exc:
        raise CsvRejected('ヘッダーをCSVとして解析できません') from exc

    if not header or all(cell == '' for cell in header):
        raise CsvRejected('有効なヘッダーがありません')

    expected_columns = len(header)
    data_rows = 0

    try:
        for record_number, row in enumerate(reader, start=2):
            if len(row) != expected_columns:
                raise CsvRejected(
                    f'レコード{record_number}の列数は{len(row)}です。'
                    f'ヘッダーの{expected_columns}列と一致しません'
                )
            data_rows += 1
    except csv.Error as exc:
        raise CsvRejected(
            f'CSV構文を解析できません(物理行{reader.line_num}付近)'
        ) from exc

    return expected_columns, data_rows


def write_new_file(path: Path, payload: bytes) -> None:
    created = False

    try:
        handle = path.open('xb')
        created = True
        with handle:
            written = handle.write(payload)
            if written != len(payload):
                raise OSError('short write')
    except FileExistsError as exc:
        raise OutputRejected(f'出力先が既に存在します: {path}') from exc
    except OSError as exc:
        if created:
            try:
                path.unlink()
            except OSError:
                pass
        raise OutputRejected(f'出力先へ書き込めません: {path}') from exc


def emit_error(code: int, message: str) -> int:
    report = {'status': 'error', 'message': message}
    print(
        json.dumps(report, ensure_ascii=False, separators=(',', ':')),
        file=sys.stderr,
    )
    return code


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description='CSVを検査し、BOMなしUTF-8へ変換します'
    )
    parser.add_argument('input', type=Path, help='入力CSV')
    parser.add_argument('output', type=Path, help='新規作成する出力CSV')
    parser.add_argument(
        '--allow-cp932',
        action='store_true',
        help='UTF-8でない入力をCP932として試します',
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()

    try:
        data = args.input.read_bytes()
    except OSError:
        return emit_error(EXIT_INPUT, f'入力を読み込めません: {args.input}')

    try:
        text, source_encoding = decode_input(data, args.allow_cp932)
    except EncodingRejected as exc:
        return emit_error(EXIT_ENCODING, str(exc))

    try:
        columns, data_rows = inspect_csv(text)
    except CsvRejected as exc:
        return emit_error(EXIT_CSV, str(exc))

    try:
        write_new_file(args.output, text.encode('utf-8'))
    except OutputRejected as exc:
        return emit_error(EXIT_OUTPUT, str(exc))

    report = {
        'status': 'ok',
        'source_encoding': source_encoding,
        'output_encoding': 'utf-8',
        'columns': columns,
        'data_rows': data_rows,
        'input': str(args.input),
        'output': str(args.output),
    }
    print(json.dumps(report, ensure_ascii=False, separators=(',', ':')))
    return EXIT_OK


if __name__ == '__main__':
    raise SystemExit(main())

再現手順

1.CP932の入力例を作る

customers_utf8.csvが存在しない作業ディレクトリで、次のコマンドを実行します。

python -c "from pathlib import Path; Path('customers_cp932.csv').write_bytes('顧客ID,氏名\r\nC001,山田太郎\r\n'.encode('cp932'))"

2.許可せずに実行する

python normalize_csv_encoding.py customers_cp932.csv customers_utf8.csv
echo $?

CP932を許可していないため、標準エラー出力と終了コードは次のようになります。出力CSVは作られません。

{"status":"error","message":"入力はUTF-8ではありません。CP932を受け入れる場合だけ --allow-cp932 を指定してください"}
3

3.入力元を確認したうえでCP932を許可する

python normalize_csv_encoding.py customers_cp932.csv customers_utf8.csv --allow-cp932
echo $?

正常時は標準出力へ1行のJSONが出ます。

{"status":"ok","source_encoding":"cp932","output_encoding":"utf-8","columns":2,"data_rows":1,"input":"customers_cp932.csv","output":"customers_utf8.csv"}
0

出力にBOMがなく、UTF-8として読めることも確認します。

python -c "from pathlib import Path; b=Path('customers_utf8.csv').read_bytes(); print(b.startswith(bytes((239, 187, 191))), b.decode('utf-8').splitlines()[0])"
False 顧客ID,氏名

4.UTF-8 BOMが複数連続する入力を拒否する

codecs.BOM_UTF8を2個付けたCSVを作り、未使用の出力先を指定します。

python -c "from pathlib import Path; import codecs; Path('double_bom.csv').write_bytes(codecs.BOM_UTF8 * 2 + b'a,b\n1,2\n')"
python normalize_csv_encoding.py double_bom.csv double_bom_output.csv
echo $?

この入力は文字コードエラーになり、出力CSVは作られません。

{"status":"error","message":"UTF-8 BOMが複数連続しているため受け入れません"}
3

5.列数が異なるCSVを拒否する

python -c "from pathlib import Path; Path('broken.csv').write_bytes(b'id,name\n1\n')"
python normalize_csv_encoding.py broken.csv unused.csv
echo $?
{"status":"error","message":"レコード2の列数は1です。ヘッダーの2列と一致しません"}
4

検査に失敗したため、unused.csvは作られません。

6.既存の出力ファイルを上書きしないことを確認する

手順3で作成したcustomers_utf8.csvを出力先にして、同じ処理をもう一度実行します。

python normalize_csv_encoding.py customers_cp932.csv customers_utf8.csv --allow-cp932
echo $?
{"status":"error","message":"出力先が既に存在します: customers_utf8.csv"}
5

xbモードで新規作成しているため、既存のcustomers_utf8.csvは変更されません。

終了コードを後続処理の条件にする

終了コード 意味 対応
0 正規化成功 出力CSVを次工程へ渡す
2 入力を読めない、または引数エラー パス、権限、引数を確認する
3 文字コードを受け入れられない 送信元の出力設定とBOMを確認する
4 CSV構文または列数が不正 ファイルを隔離して内容を確認する
5 出力できない 既存ファイル、親ディレクトリ、権限を確認する

入力ファイルの読み込みに失敗した場合は、終了コード2と1行のJSONを返します。一方、引数不足や未知のオプションをargparseが処理開始前に検出した場合は、同じ終了コード2でもJSONではなく使用法が標準エラー出力へ表示されます。--helpは使用法を表示し、終了コード0で終了します。

この実装で意図的に行わないこと

  • EUC-JPなど、契約にない文字コードの推測
  • UTF-16またはUTF-32からの自動変換
  • 複数のUTF-8 BOMを除去して処理を継続すること
  • 改行コード、区切り文字、引用符の正規化
  • 日付、金額、必須項目など業務スキーマの検証
  • 既存ファイルの上書き
  • 大容量CSVのストリーミング処理

このコードは入力全体をメモリへ読み込むため、大容量CSVには向きません。まず小さな業務ファイルの入口処理として使い、容量が大きい場合は上限を決めたうえで別の実装を検討します。

CP932としてデコードできることは、入力元が本当にCP932を採用している証明にはなりません。別の文字コードのバイト列でも偶然デコードできる可能性があるため、--allow-cp932は入力元単位で設定します。ASCII文字だけの入力はUTF-8と同じバイト表現なので、UTF-8として扱われます。

書き込み中に通常のエラーが発生した場合、コードは作成途中のファイルの削除を試みます。ただし、削除自体に失敗する場合や、プロセスの強制終了、OS障害まで含む原子的な保存は保証しません。終了コード0を確認したファイルだけを次工程へ渡します。

実務へ組み込むときの確認項目

  • 入力元ごとに、許可する文字コードを記録する
  • 下流処理には原本ではなく正規化後のパスを渡す
  • 終了コード0のときだけスキーマ移行や集計を開始する
  • 標準出力と標準エラー出力を分けて保存する
  • 原本を保管する場合は、保存期間とアクセス権を業務ルールで決める

文字コードの問題を入口で分離すれば、後段のスキーマ移行や品質検査はUTF-8だけを前提にできます。対応形式をむやみに増やさず、受け入れ条件を狭く明示することが、再現可能なCSV処理につながります。

第158回 CSV仕様変更を安全に進める:Pythonで作るv1からv2への移行スクリプト

第156回ではCSVの品質ゲートを作り、第157回ではその判定を回帰テストしました。品質ゲートで現在の仕様を固定すると、正当な列名変更や列追加も不一致として検出されます。

検査を緩めて新旧のCSVを混在させるのではなく、旧仕様から新仕様へ変換する工程を分けます。今回はCSVにスキーマバージョンを持たせ、バージョン1だけを受け付けてバージョン2を別ファイルへ出力するスクリプトを、Python標準ライブラリだけで作ります。

今回のゴール

  • CSVの新旧仕様をバージョン番号で識別する
  • 列名変更と列追加を決めた規則どおりに実行する
  • 旧CSVから判断できない値を推測で補わない
  • 主キーの欠落や重複を移行前に検出する
  • 元ファイルと既存の出力ファイルを上書きしない
  • 終了コード、成功時のJSON、出力ファイルを確認して後続処理の実行可否を判断する

新旧の仕様と移行規則を決める

例として、顧客CSVをバージョン1からバージョン2へ変更します。

バージョン1 バージョン2 移行方法
schema_version schema_version 値を1から2へ変更
customer_id customer_id 値を保持
name customer_name 列名だけを変更
email email 空欄を含めて値を保持
なし status unknownを設定

statusへ一律にactiveを設定することはできません。旧CSVだけでは、顧客が現在も有効か判断できないためです。この例では保留値のunknownを設定し、業務担当者による確認を別工程にします。

移行元CSVを用意する

次の内容をUTF-8のcustomers_v1.csvとして保存します。

schema_version,customer_id,name,email
1,C001,顧客A,contact-a@example.com
1,C002,顧客B,

この実装では、ヘッダーの名前と順序を契約の一部として扱います。列の不足、未承認の追加、並べ替えはすべてエラーになります。

一方向の移行スクリプトを実装する

次のコードをmigrate_customers.pyとして保存します。csv.DictReaderで列数の過不足を調べ、strict=Trueで一部の不正なCSV構文も検出します。これらの挙動はPythonのcsv公式ドキュメントで確認できます。

#!/usr/bin/env python3
import argparse
import csv
import json
import sys
from pathlib import Path

V1_FIELDS = ["schema_version", "customer_id", "name", "email"]
V2_FIELDS = [
    "schema_version",
    "customer_id",
    "customer_name",
    "email",
    "status",
]


class MigrationError(Exception):
    pass


def load_v1(path):
    with path.open("r", encoding="utf-8-sig", newline="") as file:
        reader = csv.DictReader(file, strict=True)
        actual_fields = reader.fieldnames

        if actual_fields != V1_FIELDS:
            raise MigrationError(
                f"v1のヘッダーではありません。"
                f"expected={V1_FIELDS}, actual={actual_fields}"
            )

        rows = []
        seen_ids = set()

        for record_no, row in enumerate(reader, start=1):
            if None in row or any(value is None for value in row.values()):
                raise MigrationError(
                    f"データレコード{record_no}: 列数がヘッダーと一致しません"
                )

            if row["schema_version"] != "1":
                raise MigrationError(
                    f"データレコード{record_no}: "
                    f"schema_versionは1である必要があります"
                )

            customer_id = row["customer_id"]
            if not customer_id.strip():
                raise MigrationError(
                    f"データレコード{record_no}: customer_idが空です"
                )

            if customer_id != customer_id.strip():
                raise MigrationError(
                    f"データレコード{record_no}: "
                    f"customer_idの前後に空白があります"
                )

            if customer_id in seen_ids:
                raise MigrationError(
                    f"データレコード{record_no}: "
                    f"customer_idが重複しています: {customer_id}"
                )

            if not row["name"].strip():
                raise MigrationError(
                    f"データレコード{record_no}: nameが空です"
                )

            seen_ids.add(customer_id)
            rows.append(row)

    if not rows:
        raise MigrationError("データ行がありません")

    return rows


def migrate_to_v2(rows):
    migrated = []

    for row in rows:
        migrated.append(
            {
                "schema_version": "2",
                "customer_id": row["customer_id"],
                "customer_name": row["name"],
                "email": row["email"],
                "status": "unknown",
            }
        )

    return migrated


def validate_v2(rows):
    seen_ids = set()

    for record_no, row in enumerate(rows, start=1):
        if list(row) != V2_FIELDS:
            raise MigrationError(
                f"データレコード{record_no}: v2の列構成が不正です"
            )

        if row["schema_version"] != "2":
            raise MigrationError(
                f"データレコード{record_no}: v2のバージョンが不正です"
            )

        customer_id = row["customer_id"]
        if not customer_id or customer_id != customer_id.strip():
            raise MigrationError(
                f"データレコード{record_no}: customer_idが不正です"
            )

        if customer_id in seen_ids:
            raise MigrationError(
                f"データレコード{record_no}: "
                f"customer_idが重複しています: {customer_id}"
            )

        if not row["customer_name"].strip():
            raise MigrationError(
                f"データレコード{record_no}: customer_nameが空です"
            )

        if row["status"] not in {"active", "inactive", "unknown"}:
            raise MigrationError(
                f"データレコード{record_no}: statusが不正です"
            )

        seen_ids.add(customer_id)


def write_v2(path, rows):
    try:
        with path.open("x", encoding="utf-8", newline="") as file:
            writer = csv.DictWriter(
                file,
                fieldnames=V2_FIELDS,
                extrasaction="raise",
            )
            writer.writeheader()
            writer.writerows(rows)
    except FileExistsError as exc:
        raise MigrationError(
            f"出力先が既に存在します: {path}"
        ) from exc


def main():
    parser = argparse.ArgumentParser(
        description="顧客CSVをスキーマv1からv2へ移行します"
    )
    parser.add_argument("source", type=Path, help="v1の入力CSV")
    parser.add_argument("output", type=Path, help="v2の出力CSV")
    args = parser.parse_args()

    try:
        if args.source.resolve() == args.output.resolve():
            raise MigrationError("入力と出力に同じパスは指定できません")

        source_rows = load_v1(args.source)
        migrated_rows = migrate_to_v2(source_rows)
        validate_v2(migrated_rows)
        write_v2(args.output, migrated_rows)
    except MigrationError as exc:
        print(f"移行失敗: {exc}", file=sys.stderr)
        return 3
    except (OSError, UnicodeError, csv.Error) as exc:
        print(f"処理エラー: {exc}", file=sys.stderr)
        return 4

    result = {
        "status": "ok",
        "source_version": "1",
        "target_version": "2",
        "rows": len(migrated_rows),
        "output": str(args.output),
    }
    print(json.dumps(result, ensure_ascii=False))
    return 0


if __name__ == "__main__":
    sys.exit(main())

実行結果を確認する

Python 3で、入力とは異なる出力先を指定します。環境によってPython 3の実行コマンド名が異なる場合は、その環境のコマンドに読み替えてください。

python3 migrate_customers.py customers_v1.csv customers_v2.csv

移行と書き込みが完了すると、標準出力に次の1行が出ます。

{"status": "ok", "source_version": "1", "target_version": "2", "rows": 2, "output": "customers_v2.csv"}

生成されるcustomers_v2.csvの内容は次のとおりです。

schema_version,customer_id,customer_name,email,status
2,C001,顧客A,contact-a@example.com,unknown
2,C002,顧客B,,unknown

入力全体と変換結果を検証してから出力を開きます。入力契約に違反している場合、出力ファイルは作られません。出力には排他的な作成モードxを使うため、同名のファイルが存在する場合も上書きせずに停止します。

終了コードと成功時の出力を確認する

終了コード 意味 対応
0 通常実行では移行とv2検証が完了。ただし、--helpも移行せずに0を返す 成功時のJSONと出力ファイルも確認する
2 引数の不足や未知の引数 実行コマンドを修正する
3 入力契約違反、同一パス、出力先の既存 CSVまたは指定先を確認する
4 ファイル、文字コード、CSV解析のエラー 入出力環境を確認する

終了コード2は、argparseが不正な引数に対して返す既定値です。一方、既定の--helpはヘルプを表示して終了コード0を返しますが、移行処理は実行しません。詳細はPythonのargparse公式ドキュメントで確認できます。

同じ出力先を指定して再実行すると、既存ファイルを保護して終了コード3を返し、次のメッセージを標準エラー出力へ出します。

移行失敗: 出力先が既に存在します: customers_v2.csv

自動処理では、終了コード0だけで移行成功と判断しません。通常の位置引数を指定して実行したうえで、終了コードが0であること、標準出力がJSONとして解析できること、statusがokであること、source_versionとtarget_versionが想定どおりであること、JSONのoutputが指定した出力先を示すこと、そのファイルが存在することを確認してから後続の品質検査へ進みます。

書き込み中に入出力エラーが起きた場合は、不完全な出力ファイルが残る可能性があります。終了コード4になったときは出力を検査し、不完全なファイルを削除または退避してから再実行します。

業務フローへ組み込む順序

  1. 新旧の列、必須条件、許容値、既定値を表にして承認する
  2. 本番データのコピーを使い、別名の出力先で移行する
  3. 終了コード0に加え、成功時のJSONと指定した出力ファイルを確認する
  4. バージョン2用の品質ゲートで件数、主キー、列構成を検査する
  5. unknownなど確認が必要な値を業務担当者が処理する
  6. 後続システムがバージョン2を読めることを確認してから入力先を切り替える
  7. 旧ファイルを社内の保存ルールに従って保管する

AIに列の対応案を作らせる場合でも、保持する値と不明値の扱いはデータの責任者が決めます。承認した規則をPythonコードに固定すれば、同じ入力から同じ結果を作り、後から点検できます。

この実装の制約

  • 入力はカンマ区切りのUTF-8またはBOM付きUTF-8に限定しています。
  • 各値は文字列として扱い、日付やメールアドレスの形式は検証しません。
  • 全行をメモリに保持するため、利用環境のメモリに収まるCSVが対象です。
  • unknownを許可しない業務では、後続システムへ渡す前に確認と更新が必要です。
  • 書き込み中の障害に備えた一時ファイル保存や原子的な切り替えは実装していません。
  • 表計算ソフトで開く場合のCSV数式インジェクション対策は含みません。必要な場合は、閲覧環境と対象列を決めたうえで別に対策します。
  • argparseの既定の--helpも終了コード0を返すため、自動処理では終了コードだけを移行成功の証拠にできません。

まとめ

CSVの仕様変更では、品質ゲートを緩めるのではなく、新旧の仕様と移行規則を分けて管理します。このスクリプトはバージョン1だけを受け付け、判断できない値をunknownとして残し、検証したバージョン2を別ファイルへ出力します。

実データのコピーで移行を試し、終了コード、成功時のJSON、出力ファイルを確認してから、生成結果をバージョン2用の品質ゲートへ渡してください。移行、検査、業務判断を分けることで、未確認の値を本番データへ混入させずに仕様を切り替えられます。

第157回 CSV品質ゲートを回帰テストする:unittestで終了コードとJSON出力を検証する

第156回では、更新前後のCSVを突合し、許可されていない変更を止める品質ゲートを作りました。しかし、品質ゲート自体を修正したときに判定が変わると、正しい更新を止めたり、契約違反の更新を通したりするおそれがあります。

今回はPython標準ライブラリだけで小さな品質ゲートを実装し、そのCLIをunittestから実行します。比較関数だけでなく、CSV入力、1行のJSON出力、標準エラー出力、終了コードまでテストします。コードはPython 3.10以上を前提とします。

今回テストする判定契約

  • 更新前後でヘッダーの名前と順序が一致すること
  • 主キーの集合が一致し、空欄や重複がないこと
  • --allowで指定した列だけが変更されていること
  • --expected-changesを指定した場合は、変更されたレコード数が一致すること
  • 処理結果を1行のJSONとして標準出力へ返し、通常時は標準エラー出力を使わないこと
終了コード 意味 後続処理
0 品質ゲートを通過 次の処理へ進める
1 CSVまたは変更内容が契約違反 更新を止めて内容を確認する
2 引数またはファイル入出力の問題 実行設定やパスを確認する

手順1:品質ゲートを実装する

次のコードをcsv_gate.pyとして保存します。UTF-8およびUTF-8 BOM付きのカンマ区切りCSVを読み込み、各フィールドを文字列のまま比較します。

from __future__ import annotations

import argparse
import csv
import json
from collections import Counter
from dataclasses import asdict, dataclass
from pathlib import Path


class GateError(Exception):
    def __init__(self, code: str, **details: object) -> None:
        super().__init__(code)
        self.code = code
        self.details = details

    def payload(self) -> dict[str, object]:
        return {
            "ok": False,
            "error": {"code": self.code, **self.details},
        }


@dataclass(frozen=True)
class GateResult:
    ok: bool
    row_count: int
    changed_keys: list[str]
    changed_columns: list[str]


def emit(payload: dict[str, object]) -> None:
    print(json.dumps(payload, ensure_ascii=False, sort_keys=True))


class JsonArgumentParser(argparse.ArgumentParser):
    def error(self, message: str) -> None:
        emit({
            "ok": False,
            "error": {"code": "usage_error", "message": message},
        })
        raise SystemExit(2)


def non_negative_int(value: str) -> int:
    try:
        number = int(value)
    except ValueError as exc:
        raise argparse.ArgumentTypeError(
            "0以上の整数を指定してください"
        ) from exc
    if number < 0:
        raise argparse.ArgumentTypeError(
            "0以上の整数を指定してください"
        )
    return number


def read_indexed_csv(
    path: Path,
    key_column: str,
) -> tuple[list[str], dict[str, dict[str, str]]]:
    reader = None
    try:
        with path.open("r", encoding="utf-8-sig", newline="") as handle:
            reader = csv.DictReader(handle, strict=True)
            headers = reader.fieldnames

            if not headers:
                raise GateError("missing_header", path=str(path))

            blank_positions = [
                index + 1
                for index, name in enumerate(headers)
                if not name.strip()
            ]
            if blank_positions:
                raise GateError(
                    "blank_header",
                    path=str(path),
                    positions=blank_positions,
                )

            duplicates = sorted(
                name
                for name, count in Counter(headers).items()
                if count > 1
            )
            if duplicates:
                raise GateError(
                    "duplicate_header",
                    path=str(path),
                    columns=duplicates,
                )

            if key_column not in headers:
                raise GateError(
                    "missing_key_column",
                    path=str(path),
                    column=key_column,
                )

            rows: dict[str, dict[str, str]] = {}
            for row in reader:
                if None in row or any(
                    value is None for value in row.values()
                ):
                    raise GateError(
                        "row_width_mismatch",
                        path=str(path),
                        line=reader.line_num,
                    )

                record_id = row[key_column]
                if not record_id.strip():
                    raise GateError(
                        "blank_key",
                        path=str(path),
                        line=reader.line_num,
                    )
                if record_id in rows:
                    raise GateError(
                        "duplicate_key",
                        path=str(path),
                        key=record_id,
                    )
                rows[record_id] = row

            return headers, rows

    except UnicodeDecodeError as exc:
        raise GateError("decode_error", path=str(path)) from exc
    except csv.Error as exc:
        line = reader.line_num if reader is not None else None
        raise GateError(
            "csv_parse_error",
            path=str(path),
            line=line,
            message=str(exc),
        ) from exc


def compare_csv(
    before_path: Path,
    after_path: Path,
    key_column: str,
    allowed_columns: list[str],
    expected_changes: int | None,
) -> GateResult:
    allowed = set(allowed_columns)
    if key_column in allowed:
        raise GateError(
            "key_column_cannot_be_allowed",
            column=key_column,
        )

    before_headers, before_rows = read_indexed_csv(
        before_path,
        key_column,
    )
    after_headers, after_rows = read_indexed_csv(
        after_path,
        key_column,
    )

    if before_headers != after_headers:
        raise GateError(
            "header_mismatch",
            before=before_headers,
            after=after_headers,
        )

    unknown_allowed = sorted(allowed - set(before_headers))
    if unknown_allowed:
        raise GateError(
            "unknown_allowed_columns",
            columns=unknown_allowed,
        )

    before_keys = set(before_rows)
    after_keys = set(after_rows)
    if before_keys != after_keys:
        raise GateError(
            "key_set_mismatch",
            added=sorted(after_keys - before_keys),
            removed=sorted(before_keys - after_keys),
        )

    changed_keys: list[str] = []
    changed_columns: set[str] = set()

    for record_id in sorted(before_rows):
        differing = [
            column
            for column in before_headers
            if column != key_column
            and before_rows[record_id][column]
            != after_rows[record_id][column]
        ]
        forbidden = sorted(set(differing) - allowed)
        if forbidden:
            raise GateError(
                "forbidden_change",
                key=record_id,
                columns=forbidden,
            )
        if differing:
            changed_keys.append(record_id)
            changed_columns.update(differing)

    if (
        expected_changes is not None
        and len(changed_keys) != expected_changes
    ):
        raise GateError(
            "unexpected_change_count",
            expected=expected_changes,
            actual=len(changed_keys),
        )

    return GateResult(
        ok=True,
        row_count=len(before_rows),
        changed_keys=changed_keys,
        changed_columns=sorted(changed_columns),
    )


def build_parser() -> argparse.ArgumentParser:
    parser = JsonArgumentParser()
    parser.add_argument("--before", required=True, type=Path)
    parser.add_argument("--after", required=True, type=Path)
    parser.add_argument("--key", required=True)
    parser.add_argument(
        "--allow",
        action="append",
        default=[],
        metavar="COLUMN",
    )
    parser.add_argument(
        "--expected-changes",
        type=non_negative_int,
    )
    return parser


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    try:
        result = compare_csv(
            before_path=args.before,
            after_path=args.after,
            key_column=args.key,
            allowed_columns=args.allow,
            expected_changes=args.expected_changes,
        )
    except GateError as exc:
        emit(exc.payload())
        return 1
    except OSError as exc:
        emit({
            "ok": False,
            "error": {
                "code": "io_error",
                "message": str(exc),
            },
        })
        return 2

    emit(asdict(result))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

CSVの形式や変更内容に関する不一致はGateErrorへ変換し、終了コード1で返します。ファイルが存在しない場合などのOSErrorは終了コード2です。引数の不足や不正な値も、JsonArgumentParserがJSONを出力して終了コード2にします。

想定外の不具合をexcept Exceptionで隠してはいません。未分類の例外は、実装を修正すべき問題としてテスト結果や実行ログに残します。

手順2:CLI全体を回帰テストする

次のコードをcsv_gate.pyと同じディレクトリへtest_csv_gate.pyとして保存します。各テストは一時ディレクトリ内にCSVを作るため、既存の業務ファイルには触れません。

from __future__ import annotations

import csv
import json
import os
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path


SCRIPT = Path(__file__).with_name("csv_gate.py").resolve()
FIELDS = ["id", "status", "note"]


class GateCliTests(unittest.TestCase):
    def setUp(self) -> None:
        self.temp_dir = tempfile.TemporaryDirectory()
        self.root = Path(self.temp_dir.name)
        self.before_rows = [
            {
                "id": "A001",
                "status": "pending",
                "note": "call, email",
            },
            {
                "id": "A002",
                "status": "pending",
                "note": "",
            },
        ]

    def tearDown(self) -> None:
        self.temp_dir.cleanup()

    def write_csv(
        self,
        name: str,
        rows: list[dict[str, str]],
        fieldnames: list[str] = FIELDS,
    ) -> Path:
        path = self.root / name
        with path.open(
            "w",
            encoding="utf-8",
            newline="",
        ) as handle:
            writer = csv.DictWriter(
                handle,
                fieldnames=fieldnames,
                lineterminator=chr(10),
            )
            writer.writeheader()
            writer.writerows(rows)
        return path

    def run_gate(
        self,
        before: Path,
        after: Path,
        allowed: tuple[str, ...] = ("status",),
        expected: int | None = None,
    ) -> tuple[
        subprocess.CompletedProcess[str],
        dict[str, object],
    ]:
        command = [
            sys.executable,
            str(SCRIPT),
            "--before",
            str(before),
            "--after",
            str(after),
            "--key",
            "id",
        ]
        for column in allowed:
            command.extend(["--allow", column])
        if expected is not None:
            command.extend([
                "--expected-changes",
                str(expected),
            ])

        child_env = os.environ.copy()
        child_env["PYTHONIOENCODING"] = "utf-8"

        completed = subprocess.run(
            command,
            capture_output=True,
            text=True,
            encoding="utf-8",
            check=False,
            timeout=10,
            env=child_env,
        )

        self.assertEqual(completed.stderr, "", completed.stderr)
        stdout_lines = completed.stdout.splitlines()
        self.assertEqual(
            len(stdout_lines),
            1,
            completed.stdout,
        )
        return completed, json.loads(stdout_lines[0])

    def test_allowed_change_passes(self) -> None:
        before = self.write_csv(
            "before.csv",
            self.before_rows,
        )
        after = self.write_csv("after.csv", [
            {
                "id": "A001",
                "status": "approved",
                "note": "call, email",
            },
            {
                "id": "A002",
                "status": "pending",
                "note": "",
            },
        ])

        completed, payload = self.run_gate(
            before,
            after,
            expected=1,
        )

        self.assertEqual(completed.returncode, 0)
        self.assertEqual(payload, {
            "ok": True,
            "row_count": 2,
            "changed_keys": ["A001"],
            "changed_columns": ["status"],
        })

    def test_header_mismatch_is_rejected(self) -> None:
        before = self.write_csv(
            "header_before.csv",
            self.before_rows,
        )
        after = self.write_csv(
            "header_after.csv",
            self.before_rows,
            fieldnames=["id", "note", "status"],
        )

        completed, payload = self.run_gate(before, after)

        self.assertEqual(completed.returncode, 1)
        self.assertEqual(
            payload["error"]["code"],
            "header_mismatch",
        )

    def test_rejected_cases(self) -> None:
        cases = [
            (
                "forbidden",
                [
                    {
                        "id": "A001",
                        "status": "approved",
                        "note": "call only",
                    },
                    {
                        "id": "A002",
                        "status": "pending",
                        "note": "",
                    },
                ],
                "forbidden_change",
            ),
            (
                "duplicate",
                [
                    {
                        "id": "A001",
                        "status": "approved",
                        "note": "call, email",
                    },
                    {
                        "id": "A001",
                        "status": "pending",
                        "note": "",
                    },
                ],
                "duplicate_key",
            ),
            (
                "blank",
                [
                    {
                        "id": " ",
                        "status": "approved",
                        "note": "call, email",
                    },
                    {
                        "id": "A002",
                        "status": "pending",
                        "note": "",
                    },
                ],
                "blank_key",
            ),
            (
                "removed",
                [
                    {
                        "id": "A001",
                        "status": "approved",
                        "note": "call, email",
                    },
                ],
                "key_set_mismatch",
            ),
        ]

        for name, after_rows, expected_code in cases:
            with self.subTest(name=name):
                before = self.write_csv(
                    f"{name}_before.csv",
                    self.before_rows,
                )
                after = self.write_csv(
                    f"{name}_after.csv",
                    after_rows,
                )

                completed, payload = self.run_gate(
                    before,
                    after,
                )

                self.assertEqual(completed.returncode, 1)
                self.assertEqual(
                    payload["error"]["code"],
                    expected_code,
                )

    def test_wrong_expected_change_count_is_rejected(
        self,
    ) -> None:
        before = self.write_csv(
            "count_before.csv",
            self.before_rows,
        )
        after = self.write_csv("count_after.csv", [
            {
                "id": "A001",
                "status": "approved",
                "note": "call, email",
            },
            {
                "id": "A002",
                "status": "pending",
                "note": "",
            },
        ])

        completed, payload = self.run_gate(
            before,
            after,
            expected=2,
        )

        self.assertEqual(completed.returncode, 1)
        self.assertEqual(
            payload["error"]["code"],
            "unexpected_change_count",
        )

    def test_missing_file_uses_exit_code_2(self) -> None:
        missing = self.root / "missing.csv"
        after = self.write_csv(
            "io_after.csv",
            self.before_rows,
        )

        completed, payload = self.run_gate(missing, after)

        self.assertEqual(completed.returncode, 2)
        self.assertEqual(
            payload["error"]["code"],
            "io_error",
        )

    def test_invalid_argument_uses_exit_code_2(self) -> None:
        before = self.write_csv(
            "argument_before.csv",
            self.before_rows,
        )
        after = self.write_csv(
            "argument_after.csv",
            self.before_rows,
        )

        completed, payload = self.run_gate(
            before,
            after,
            expected=-1,
        )

        self.assertEqual(completed.returncode, 2)
        self.assertEqual(
            payload["error"]["code"],
            "usage_error",
        )


if __name__ == "__main__":
    unittest.main()

run_gateは、標準出力が1行だけであることと、標準エラー出力が空であることを確認してからJSONを解析します。CLIが例外終了した場合や、JSON以外の文字列を出した場合もテスト失敗になります。

子プロセスにはPYTHONIOENCODING=utf-8を渡しています。これにより、OSの既定文字コードに左右されず、日本語を含むJSONをUTF-8として読み取れます。

追加ライブラリは不要です。2つのファイルを保存したディレクトリで、次のコマンドを実行します。

python -m unittest -v test_csv_gate.py

6件のテストがすべてokになり、最後にOKと表示されれば、正常系、主要な契約違反、変更件数の不一致、入出力エラー、不正な引数を確認できています。実行時間などの表示は環境によって異なります。

手順3:実際のCSVで入出力を確認する

更新前の内容をbefore.csvとして保存します。

id,status,note
A001,pending,"call, email"
A002,pending,

更新後のafter.csvでは、許可対象のstatusだけを変更します。

id,status,note
A001,approved,"call, email"
A002,pending,

変更可能列と予定変更件数を指定して実行します。

python csv_gate.py --before before.csv --after after.csv --key id --allow status --expected-changes 1

通過時の標準出力は次の1行で、終了コードは0です。

{"changed_columns": ["status"], "changed_keys": ["A001"], "ok": true, "row_count": 2}

A001のnoteもcall onlyへ変更すると、許可されていない列の変更として終了コード1になります。

{"error": {"code": "forbidden_change", "columns": ["note"], "key": "A001"}, "ok": false}

このテストで守れる範囲

subprocess.runでCLIを別プロセスとして呼び出すため、引数の解釈、CSVの読み込み、JSONの形式、標準出力と標準エラー出力、終了コードをまとめて検査できます。後続の処理が依存するCLIの境界を確認するテストです。

一方、次の制約は残ります。

  • 値は文字列として厳密に比較します。01と1、末尾空白の有無は別の値です。
  • 引用符や改行コードだけが異なり、CSVから読み取ったフィールド値が同じ場合は変更に数えません。
  • 列の追加、削除、並べ替えと、レコードの追加、削除はすべて拒否します。
  • 更新前後のCSVを辞書としてメモリへ保持します。大容量データでは、キー順に整列した逐次比較やデータベースを使う別の設計が必要です。
  • --expected-changesが数えるのはセル数ではなく、1つ以上の値が変わったレコード数です。
  • --allowの指定が業務上正しいかどうかは判定できません。承認済みの変更内容に基づき、人または管理された実行設定から明示的に渡す必要があります。
  • timeout=10は小さなテスト用CSVを前提とした停止策です。実データの処理時間を保証する設定ではありません。

次の一歩

実際の列名に合わせて--keyと--allowを変更し、正常な更新、許可外の列変更、空の主キー、重複キー、レコード欠落のテストデータを用意します。品質ゲートの仕様を変更するときは、期待する動作を先にテストへ反映し、そのテストが意図どおり失敗することを確かめてから実装を変更すると、判定の変化を追いやすくなります。

第156回 更新後のCSVを機械的に突合する:Pythonで作る件数・キー・変更列の品質ゲート

CSV更新処理がエラーなく終了しても、想定した範囲だけが変更されたとは限りません。対象外の金額列が書き換わった、行が欠落した、承認した件数と実際の変更件数が違う、といった問題は終了コードだけでは検出できません。

第155回では、変更案を確認してから適用するPlan/Apply型の更新を扱いました。今回はその次の段階として、更新前のCSVと適用後の候補CSVを突合し、後続処理へ進めてよいかを機械的に判定します。

今回作る品質ゲート

例では、注文データのstatusとreview_noteだけを更新できるものとします。Pythonスクリプトは次の条件を確認します。

  1. 更新前後で列名と列順が変わっていない
  2. 主キーの追加や欠落がない
  3. 主キーが空欄や重複になっていない
  4. 許可した列以外が変更されていない
  5. 更新前後の両方に存在するキーの変更行数が、承認済みの予定件数と一致する

これは行の追加や削除を行わない「既存行の更新専用」ルールです。追加・削除を含む業務では、この合格条件をそのまま使用できません。また、許可列に書かれた値の正しさまでは判定しません。

1.検証用CSVを用意する

Python 3.10以上を使用します。外部ライブラリは不要です。次の3ファイルを同じディレクトリへ保存してください。

更新前:before.csv

order_id,customer,amount,status,review_note
A001,青木商店,12000,pending,
A002,北川企画,8500,pending,
A003,南工業,5000,approved,確認済み

正しい更新例:after_ok.csv

order_id,customer,amount,status,review_note
A001,青木商店,12000,approved,担当者確認済み
A002,北川企画,8500,pending,
A003,南工業,5000,approved,確認済み

A001のstatusとreview_noteだけが変わっています。変更されたセルは2つですが、変更行数は1行です。

誤った更新例:after_ng.csv

order_id,customer,amount,status,review_note
A001,青木商店,12000,approved,担当者確認済み
A002,北川企画,9500,pending,
A004,東商事,5000,approved,確認済み

こちらには、許可していないamountの変更と、A003の欠落、A004の追加が含まれています。

2.突合スクリプトを作る

次のコードをreconcile_csv.pyとして保存します。レポートには変更前後の値を含めず、追加キー、欠落キー、変更行、禁止列の変更を各一覧につき最大20件記録します。

from __future__ import annotations

import argparse
import csv
import json
import sys
from collections import Counter
from pathlib import Path

SAMPLE_LIMIT = 20


class CsvInputError(Exception):
    pass


def read_snapshot(
    path: Path, key: str
) -> tuple[list[str], dict[str, dict[str, str]]]:
    try:
        with path.open("r", encoding="utf-8-sig", newline="") as handle:
            reader = csv.DictReader(handle, strict=True)
            if reader.fieldnames is None:
                raise CsvInputError(f"{path}: ヘッダーがありません")

            headers = reader.fieldnames
            duplicate_names = [
                name
                for name, count in Counter(headers).items()
                if count > 1
            ]

            if any(name.strip() == "" for name in headers):
                raise CsvInputError(f"{path}: 空の列名があります")
            if duplicate_names:
                names = ", ".join(sorted(duplicate_names))
                raise CsvInputError(
                    f"{path}: 列名が重複しています: {names}"
                )
            if key not in headers:
                raise CsvInputError(
                    f"{path}: キー列 {key!r} がありません"
                )

            rows: dict[str, dict[str, str]] = {}
            for row in reader:
                line = reader.line_num

                if None in row:
                    raise CsvInputError(
                        f"{path}:{line}: ヘッダーより多い値があります"
                    )

                missing = [
                    name for name, value in row.items() if value is None
                ]
                if missing:
                    names = ", ".join(missing)
                    raise CsvInputError(
                        f"{path}:{line}: 値が不足している列: {names}"
                    )

                record_key = row[key]
                if record_key.strip() == "":
                    raise CsvInputError(
                        f"{path}:{line}: キー列 {key!r} が空です"
                    )
                if record_key != record_key.strip():
                    raise CsvInputError(
                        f"{path}:{line}: キーの前後に空白があります"
                    )
                if record_key in rows:
                    raise CsvInputError(
                        f"{path}:{line}: キー {record_key!r} が重複しています"
                    )

                rows[record_key] = row

    except CsvInputError:
        raise
    except (OSError, UnicodeError, csv.Error) as exc:
        raise CsvInputError(f"{path}: {exc}") from exc

    return headers, rows


def reconcile(
    before_headers: list[str],
    before_rows: dict[str, dict[str, str]],
    after_headers: list[str],
    after_rows: dict[str, dict[str, str]],
    key: str,
    allowed: set[str],
    expected_changed: int | None,
) -> dict[str, object]:
    before_columns = set(before_headers)
    after_columns = set(after_headers)
    header_added = sorted(after_columns - before_columns)
    header_removed = sorted(before_columns - after_columns)
    header_order_changed = (
        not header_added
        and not header_removed
        and before_headers != after_headers
    )

    before_keys = set(before_rows)
    after_keys = set(after_rows)
    added_keys = after_keys - before_keys
    removed_keys = before_keys - after_keys
    common_keys = sorted(before_keys & after_keys)

    change_counts: Counter[str] = Counter()
    changed_samples: list[dict[str, object]] = []
    forbidden_samples: list[dict[str, object]] = []
    changed_rows = 0
    forbidden_rows = 0
    forbidden_cells = 0

    for record_key in common_keys:
        changed = [
            column
            for column in before_headers
            if column != key
            and column in after_columns
            and before_rows[record_key][column]
            != after_rows[record_key][column]
        ]
        if not changed:
            continue

        changed_rows += 1
        change_counts.update(changed)

        if len(changed_samples) < SAMPLE_LIMIT:
            changed_samples.append(
                {"key": record_key, "columns": changed}
            )

        forbidden = [
            column for column in changed if column not in allowed
        ]
        if forbidden:
            forbidden_rows += 1
            forbidden_cells += len(forbidden)
            if len(forbidden_samples) < SAMPLE_LIMIT:
                forbidden_samples.append(
                    {"key": record_key, "columns": forbidden}
                )

    violations: list[dict[str, object]] = []

    if before_headers != after_headers:
        violations.append(
            {
                "rule": "same_header",
                "added": header_added,
                "removed": header_removed,
                "order_changed": header_order_changed,
            }
        )

    if added_keys or removed_keys:
        violations.append(
            {
                "rule": "same_key_set",
                "added_count": len(added_keys),
                "removed_count": len(removed_keys),
            }
        )

    if forbidden_cells:
        violations.append(
            {
                "rule": "allowed_columns_only",
                "row_count": forbidden_rows,
                "cell_count": forbidden_cells,
            }
        )

    if (
        expected_changed is not None
        and changed_rows != expected_changed
    ):
        violations.append(
            {
                "rule": "expected_changed_rows",
                "expected": expected_changed,
                "actual": changed_rows,
            }
        )

    return {
        "ok": not violations,
        "kind": "reconciliation",
        "policy": {
            "key": key,
            "allowed_columns": sorted(allowed),
            "expected_changed_rows": expected_changed,
        },
        "counts": {
            "before_rows": len(before_rows),
            "after_rows": len(after_rows),
            "common_rows": len(common_keys),
            "added_keys": len(added_keys),
            "removed_keys": len(removed_keys),
            "changed_rows": changed_rows,
            "unchanged_rows": len(common_keys) - changed_rows,
            "forbidden_changed_rows": forbidden_rows,
            "forbidden_changed_cells": forbidden_cells,
        },
        "change_counts_by_column": dict(sorted(change_counts.items())),
        "samples": {
            "added_keys": sorted(added_keys)[:SAMPLE_LIMIT],
            "removed_keys": sorted(removed_keys)[:SAMPLE_LIMIT],
            "changed_rows": changed_samples,
            "forbidden_changes": forbidden_samples,
        },
        "violations": violations,
    }


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="更新前後のCSVを突合します"
    )
    parser.add_argument("--before", required=True, type=Path)
    parser.add_argument("--after", required=True, type=Path)
    parser.add_argument("--key", required=True)
    parser.add_argument("--allow", action="append", default=[])
    parser.add_argument("--expect-changed", type=int)
    parser.add_argument("--report", required=True, type=Path)
    args = parser.parse_args()

    if args.expect_changed is not None and args.expect_changed < 0:
        parser.error("--expect-changedには0以上を指定してください")

    before_path = args.before.resolve()
    after_path = args.after.resolve()
    report_path = args.report.resolve()

    if before_path == after_path:
        parser.error("--beforeと--afterには別のファイルを指定してください")
    if report_path in {before_path, after_path}:
        parser.error("--reportに入力CSVと同じパスは指定できません")

    return args


def write_report(path: Path, report: dict[str, object]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    payload = json.dumps(
        report, ensure_ascii=False, indent=2
    ) + "\n"
    path.write_text(payload, encoding="utf-8")


def main() -> int:
    args = parse_args()

    try:
        before_headers, before_rows = read_snapshot(
            args.before, args.key
        )
        after_headers, after_rows = read_snapshot(
            args.after, args.key
        )

        allowed = set(args.allow)
        if args.key in allowed:
            raise CsvInputError(
                "キー列は--allowに指定できません"
            )

        unknown = sorted(allowed - set(before_headers))
        if unknown:
            names = ", ".join(unknown)
            raise CsvInputError(
                f"--allowに存在しない列があります: {names}"
            )

        result = reconcile(
            before_headers,
            before_rows,
            after_headers,
            after_rows,
            args.key,
            allowed,
            args.expect_changed,
        )
        exit_code = 0 if result["ok"] else 1

    except CsvInputError as exc:
        result = {
            "ok": False,
            "kind": "input_error",
            "message": str(exc),
        }
        exit_code = 2

    try:
        write_report(args.report, result)
    except OSError as exc:
        print(
            f"ERROR: レポートを書き込めません: {exc}",
            file=sys.stderr,
        )
        return 2

    if exit_code == 2:
        message = result["message"]
        print(f"ERROR: {message}", file=sys.stderr)
    else:
        status = "PASS" if result["ok"] else "FAIL"
        counts = result["counts"]
        changed_rows = counts["changed_rows"]
        violation_count = len(result["violations"])
        print(
            f"{status}: changed_rows={changed_rows}, "
            f"violations={violation_count}"
        )

    return exit_code


if __name__ == "__main__":
    sys.exit(main())

3.正しい更新を検証する

--allowは変更を許可する列ごとに指定します。承認済みの変更対象が1行なら、--expect-changed 1を付けます。

python3 reconcile_csv.py --before before.csv --after after_ok.csv --key order_id --allow status --allow review_note --expect-changed 1 --report reconciliation_ok.json
echo $?

bashなどのシェルでは、標準出力と終了コードは次のようになります。

PASS: changed_rows=1, violations=0
0

reconciliation_ok.jsonには次のレポートが出力されます。

{
  "ok": true,
  "kind": "reconciliation",
  "policy": {
    "key": "order_id",
    "allowed_columns": [
      "review_note",
      "status"
    ],
    "expected_changed_rows": 1
  },
  "counts": {
    "before_rows": 3,
    "after_rows": 3,
    "common_rows": 3,
    "added_keys": 0,
    "removed_keys": 0,
    "changed_rows": 1,
    "unchanged_rows": 2,
    "forbidden_changed_rows": 0,
    "forbidden_changed_cells": 0
  },
  "change_counts_by_column": {
    "review_note": 1,
    "status": 1
  },
  "samples": {
    "added_keys": [],
    "removed_keys": [],
    "changed_rows": [
      {
        "key": "A001",
        "columns": [
          "status",
          "review_note"
        ]
      }
    ],
    "forbidden_changes": []
  },
  "violations": []
}

4.誤った更新を検出する

同じ条件でafter_ng.csvを検証します。

python3 reconcile_csv.py --before before.csv --after after_ng.csv --key order_id --allow status --allow review_note --expect-changed 1 --report reconciliation_ng.json
echo $?

結果は不合格です。

FAIL: changed_rows=2, violations=3
1

changed_rowsは更新前後の両方に存在するキーだけを比較した件数です。この例ではA001とA002が該当します。A004の追加とA003の欠落は、変更行数ではなく1件のキー集合違反として記録されます。これに、A002のamount変更と予定件数の不一致を加えた合計3件の違反です。

レポートには変更前後の値を記録しないため、金額の値そのものは含まれません。ただし、主キーにも機密性がある場合は、サンプルを記録しない実装へ変更するか、レポートをアクセス制御された場所へ保存してください。

終了コードを後続処理の条件にする

終了コード 意味 後続処理
0 このスクリプトで設定した突合条件に合格 必要な値検証や人の確認を行い、次工程へ進める
1 比較結果が業務ルールに違反 候補CSVを採用せず、レポートを確認する
2 引数、入力CSV、レポート書き込みなどのエラー 実行条件や入力ファイルを修正する

ファイル不足、文字コードエラー、CSV解析エラー、列数の不一致、重複キーなど、比較を続けられない問題は終了コード2になります。CSVを比較できたものの、許可されていない変更などが見つかった場合は終了コード1です。これにより、実行条件を修正すべきエラーと、担当者が差分を確認すべき不一致を分けられます。

入力処理中に検出したエラーでは、kindがinput_errorのJSONレポートも出力します。ただし、必須引数の欠落などをargparseが検出した場合は、レポート作成前に終了コード2で停止します。レポートの書き込みに失敗した場合も終了コード2になりますが、指定先を直接上書きする実装のため、完全なJSONレポートが作成されたとは限りません。書き込み失敗の状況によっては、古いファイル、空のファイル、または途中まで書き込まれたファイルが残る可能性があります。後続処理はレポートの存在だけで成功を判断せず、必ず今回のプロセスの終了コードを確認してください。

AIを含む更新フローへの組み込み方

  1. AIまたは担当者が変更案を作る
  2. 元CSVを上書きせず、候補CSVへ適用する
  3. 承認時の予定件数と許可列を引数にして突合する
  4. 終了コードとJSONレポートを確認し、必要な値検証や人の確認を終えてから正式ファイルへ反映する

AIは分類や更新案の作成に使えますが、このスクリプトの合否は列、キー集合、変更範囲、件数という明示的な規則で判定します。終了コード0は、更新内容そのものが正しいという保証ではありません。

この実装の制約

  • 前後のCSVを辞書としてメモリへ読み込むため、利用可能メモリに収まるファイルが対象です。
  • 値は文字列として比較します。12000と12000.0は異なる値です。数値や日付を意味で比較したい場合は、比較前の正規化が必要です。
  • 許可列に入っていれば、誤った値でも合格する可能性があります。状態値の候補や金額範囲などは、別の値検証ルールで補います。
  • --expect-changedが確認するのは、更新前後の両方に存在するキーの変更行数だけです。承認対象とは別の行が同じ件数だけ変更された場合までは検出できません。対象キーまで承認している運用では、承認済みキー集合との照合を追加する必要があります。
  • 行追加や削除を伴う処理では、キー集合一致のルールを業務仕様に合わせて変更する必要があります。
  • --expect-changedを省略すると変更行数は合否条件になりません。承認済み件数がある運用では明示的に指定します。
  • この実装は列順の変更も不合格にします。後続処理が列名だけを参照し、列順を問わない場合はルールを緩和できます。
  • JSONレポートは指定先へ直接上書きします。書き込み中の容量不足や強制終了などが起きると、古いファイル、空のファイル、または不完全なファイルが残る可能性があります。この影響を減らすには、一時ファイルへ書き込み、書き込み完了後に置き換える方式を検討してください。

参考にした公式資料

まとめ

安全なCSV更新は、処理が完了した時点では終わりません。更新前のスナップショットと候補CSVを突合し、列、キー集合、変更範囲、件数を確認することで、想定外の構造変更や更新を検出できます。まずは検証用CSVで終了コード0と1を確認し、入力ファイル名を誤らせるなどして終了コード2の扱いも確かめてください。そのうえで、値の妥当性確認や人の承認と組み合わせ、実際の更新処理へ品質ゲートとして組み込みます。

第155回 AIの提案を直接反映しない:Pythonで作るドライランとPlan/Apply型CSV更新

AIに顧客情報や申請データを分類させる場合、結果をそのまま元データへ書き込むと、誤分類や入力の取り違えを後から確認しにくくなります。そこで今回は、AIや担当者が作った更新案をいったん「変更計画」に変換し、人が確認してから別のCSVへ適用する仕組みを作ります。

前回扱った増分処理と冪等性が「同じ処理を安全に再実行する設計」だとすれば、今回の主題は「提案と適用の間に確認可能な境界を置く設計」です。生成AIのAPI自体は呼び出さないため、APIキーなしで再現できます。

今回作るワークフロー

  1. 現在のデータをrecords.csvに用意する。
  2. AIまたは担当者の更新案をsuggestions.csvとして受け取る。
  3. planコマンドで入力を検証し、plan.jsonを作る。この段階では元データを変更しない。
  4. 担当者が変更内容を確認し、計画ファイルのSHA-256ハッシュを取得する。
  5. applyコマンドでハッシュと元データの一致を再確認し、更新結果を新しいCSVへ出力する。

適用時に元CSVが計画作成時と一致しない場合、または計画ファイルが承認ハッシュ取得時と一致しない場合、適用処理は失敗します。元CSVは上書きせず、出力先がすでに存在する場合も停止します。

前提と対象範囲

  • Python 3.10以上を使用します。外部ライブラリは不要です。
  • 例ではpending、approved、rejectedの3状態だけを許可します。
  • 入力をメモリへ読み込むため、担当者がレビューできる規模のCSVを対象とします。
  • 同じ出力先へ複数のプロセスが同時に書き込まない運用を前提とします。

入力CSVを用意する

records.csvは現在の正本です。

record_id,name,status
A001,青木商店,pending
A002,北川企画,pending
A003,南製作所,approved

suggestions.csvはAIや担当者が作った提案です。理由も必須にして、レビュー時に判断根拠を確認できるようにします。

record_id,proposed_status,reason
A001,approved,必要項目を確認済み
A002,rejected,連絡先が不足
A003,approved,現状維持

この例ではA001とA002が変更対象です。A003は現在値と提案値が同じなので、計画には変更操作として追加しません。

Plan/Applyを実装する

次のコードをworkflow.pyとして保存します。CSVの重複ID、未知のID、列不足、不正な状態、空の理由を計画作成前に検出します。

from __future__ import annotations

import argparse
import csv
import hashlib
import io
import json
import os
import sys
import tempfile
from pathlib import Path
from typing import Any

ALLOWED_STATUSES = {'pending', 'approved', 'rejected'}
PLAN_VERSION = 1


class WorkflowError(Exception):
    pass


def sha256_bytes(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def parse_csv(
    data: bytes,
    label: str,
    required: set[str],
) -> tuple[list[str], list[dict[str, str]]]:
    try:
        text = data.decode('utf-8-sig')
    except UnicodeDecodeError as exc:
        raise WorkflowError(f'{label}: UTF-8として読み込めません') from exc

    reader = csv.DictReader(io.StringIO(text, newline=''), strict=True)
    if not reader.fieldnames:
        raise WorkflowError(f'{label}: ヘッダーがありません')

    fields = list(reader.fieldnames)
    if any(name == '' for name in fields):
        raise WorkflowError(f'{label}: 空の列名があります')
    if len(fields) != len(set(fields)):
        raise WorkflowError(f'{label}: 列名が重複しています')

    missing = sorted(required - set(fields))
    if missing:
        missing_text = ', '.join(missing)
        raise WorkflowError(f'{label}: 必須列がありません: {missing_text}')

    rows: list[dict[str, str]] = []
    for data_no, row in enumerate(reader, start=1):
        if None in row or any(value is None for value in row.values()):
            raise WorkflowError(f'{label}: データ行{data_no}の列数が不正です')
        rows.append(dict(row))
    return fields, rows


def load_records(
    data: bytes,
    label: str,
) -> tuple[list[str], list[dict[str, str]], dict[str, dict[str, str]]]:
    fields, rows = parse_csv(data, label, {'record_id', 'status'})
    index: dict[str, dict[str, str]] = {}

    for data_no, row in enumerate(rows, start=1):
        record_id = row['record_id']
        status = row['status']

        if not record_id or record_id != record_id.strip():
            raise WorkflowError(f'{label}: データ行{data_no}のrecord_idが不正です')
        if record_id in index:
            raise WorkflowError(f'{label}: record_idが重複しています: {record_id}')
        if status not in ALLOWED_STATUSES:
            raise WorkflowError(f'{label}: {record_id}のstatusが不正です: {status}')

        index[record_id] = row
    return fields, rows, index


def atomic_write_text(
    path: Path,
    text: str,
    replace: bool = False,
) -> None:
    if not path.parent.is_dir():
        raise WorkflowError(f'出力先ディレクトリがありません: {path.parent}')
    if path.exists() and not replace:
        raise WorkflowError(f'出力先がすでに存在します: {path}')

    temp_path: Path | None = None
    try:
        with tempfile.NamedTemporaryFile(
            mode='w',
            encoding='utf-8',
            newline='',
            dir=path.parent,
            prefix=f'.{path.name}.',
            suffix='.tmp',
            delete=False,
        ) as temporary:
            temp_path = Path(temporary.name)
            temporary.write(text)
            temporary.flush()
            os.fsync(temporary.fileno())

        if path.exists() and not replace:
            raise WorkflowError(f'出力先がすでに存在します: {path}')
        os.replace(temp_path, path)
    except BaseException:
        if temp_path is not None:
            try:
                temp_path.unlink(missing_ok=True)
            except OSError:
                pass
        raise


def make_plan(
    records_path: Path,
    suggestions_path: Path,
    plan_path: Path,
    replace: bool,
) -> dict[str, Any]:
    if plan_path.resolve() in {
        records_path.resolve(),
        suggestions_path.resolve(),
    }:
        raise WorkflowError('計画ファイルを入力CSVと同じ場所にはできません')

    records_bytes = records_path.read_bytes()
    suggestions_bytes = suggestions_path.read_bytes()
    _, record_rows, records = load_records(records_bytes, str(records_path))
    _, suggestions = parse_csv(
        suggestions_bytes,
        str(suggestions_path),
        {'record_id', 'proposed_status', 'reason'},
    )

    seen: set[str] = set()
    operations: list[dict[str, str]] = []
    unchanged = 0

    for data_no, suggestion in enumerate(suggestions, start=1):
        record_id = suggestion['record_id']
        proposed = suggestion['proposed_status']
        reason = suggestion['reason'].strip()

        if not record_id or record_id != record_id.strip():
            raise WorkflowError(
                f'{suggestions_path}: データ行{data_no}のrecord_idが不正です'
            )
        if record_id in seen:
            raise WorkflowError(f'提案のrecord_idが重複しています: {record_id}')
        seen.add(record_id)

        if record_id not in records:
            raise WorkflowError(f'元データに存在しないrecord_idです: {record_id}')
        if proposed not in ALLOWED_STATUSES:
            raise WorkflowError(f'{record_id}の提案状態が不正です: {proposed}')
        if not reason:
            raise WorkflowError(f'{record_id}のreasonが空です')

        current = records[record_id]['status']
        if current == proposed:
            unchanged += 1
            continue

        operations.append(
            {
                'record_id': record_id,
                'before': current,
                'after': proposed,
                'reason': reason,
            }
        )

    operations.sort(key=lambda operation: operation['record_id'])
    plan: dict[str, Any] = {
        'format_version': PLAN_VERSION,
        'records_sha256': sha256_bytes(records_bytes),
        'suggestions_sha256': sha256_bytes(suggestions_bytes),
        'record_count': len(record_rows),
        'suggestion_count': len(suggestions),
        'unchanged_count': unchanged,
        'operations': operations,
    }

    plan_text = json.dumps(
        plan,
        ensure_ascii=False,
        sort_keys=True,
        indent=2,
    ) + '\n'
    atomic_write_text(plan_path, plan_text, replace=replace)
    return plan


def load_plan(data: bytes, label: str) -> dict[str, Any]:
    try:
        parsed = json.loads(data.decode('utf-8-sig'))
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise WorkflowError(f'{label}: 有効なJSONではありません') from exc

    if not isinstance(parsed, dict):
        raise WorkflowError(f'{label}: JSONオブジェクトではありません')
    if type(parsed.get('format_version')) is not int:
        raise WorkflowError(f'{label}: format_versionが不正です')
    if parsed['format_version'] != PLAN_VERSION:
        raise WorkflowError(f'{label}: 未対応のformat_versionです')

    source_hash = parsed.get('records_sha256')
    if (
        not isinstance(source_hash, str)
        or len(source_hash) != 64
        or any(character not in '0123456789abcdef' for character in source_hash.lower())
    ):
        raise WorkflowError(f'{label}: records_sha256が不正です')
    if not isinstance(parsed.get('operations'), list):
        raise WorkflowError(f'{label}: operationsが配列ではありません')
    return parsed


def apply_plan(
    records_path: Path,
    plan_path: Path,
    output_path: Path,
    approved_hash: str,
) -> dict[str, Any]:
    if output_path.resolve() in {records_path.resolve(), plan_path.resolve()}:
        raise WorkflowError('出力CSVを入力CSVまたは計画ファイルと同じ場所にはできません')

    approved = approved_hash.strip().lower()
    if (
        len(approved) != 64
        or any(character not in '0123456789abcdef' for character in approved)
    ):
        raise WorkflowError('承認ハッシュは64桁の16進数で指定してください')

    plan_bytes = plan_path.read_bytes()
    actual_plan_hash = sha256_bytes(plan_bytes)
    if actual_plan_hash != approved:
        raise WorkflowError('計画ファイルが承認時の内容と一致しません')
    plan = load_plan(plan_bytes, str(plan_path))

    records_bytes = records_path.read_bytes()
    if sha256_bytes(records_bytes) != plan['records_sha256'].lower():
        raise WorkflowError(
            '入力CSVが計画作成後に変化しています。計画を作り直してください'
        )

    fields, rows, records = load_records(records_bytes, str(records_path))
    seen: set[str] = set()

    for operation_no, operation in enumerate(plan['operations'], start=1):
        if not isinstance(operation, dict):
            raise WorkflowError(f'操作{operation_no}がオブジェクトではありません')

        required = {'record_id', 'before', 'after', 'reason'}
        if not required.issubset(operation):
            raise WorkflowError(f'操作{operation_no}の項目が不足しています')
        if not all(isinstance(operation[key], str) for key in required):
            raise WorkflowError(f'操作{operation_no}に文字列以外の値があります')

        record_id = operation['record_id']
        before = operation['before']
        after = operation['after']

        if record_id in seen:
            raise WorkflowError(f'計画内のrecord_idが重複しています: {record_id}')
        seen.add(record_id)
        if record_id not in records:
            raise WorkflowError(f'計画対象が元データにありません: {record_id}')
        if before not in ALLOWED_STATUSES or after not in ALLOWED_STATUSES:
            raise WorkflowError(f'{record_id}の状態が不正です')
        if before == after:
            raise WorkflowError(f'{record_id}は変更前後が同じです')
        if not operation['reason'].strip():
            raise WorkflowError(f'{record_id}のreasonが空です')
        if records[record_id]['status'] != before:
            raise WorkflowError(f'{record_id}の変更前状態が一致しません')

        records[record_id]['status'] = after

    buffer = io.StringIO(newline='')
    writer = csv.DictWriter(buffer, fieldnames=fields, lineterminator='\n')
    writer.writeheader()
    writer.writerows(rows)
    atomic_write_text(output_path, buffer.getvalue())

    return {
        'applied': len(plan['operations']),
        'output': str(output_path),
        'plan_sha256': actual_plan_hash,
    }


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description='CSVの状態変更を計画し、承認後に別ファイルへ適用します'
    )
    commands = parser.add_subparsers(dest='command', required=True)

    plan_command = commands.add_parser('plan', help='変更計画を作成します')
    plan_command.add_argument('--records', type=Path, required=True)
    plan_command.add_argument('--suggestions', type=Path, required=True)
    plan_command.add_argument('--plan', type=Path, required=True)
    plan_command.add_argument('--replace-plan', action='store_true')

    apply_command = commands.add_parser('apply', help='承認済み計画を適用します')
    apply_command.add_argument('--records', type=Path, required=True)
    apply_command.add_argument('--plan', type=Path, required=True)
    apply_command.add_argument('--output', type=Path, required=True)
    apply_command.add_argument('--approve-hash', required=True)
    return parser


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    try:
        if args.command == 'plan':
            plan = make_plan(
                args.records,
                args.suggestions,
                args.plan,
                replace=args.replace_plan,
            )
            result = {
                'mode': 'plan',
                'plan': str(args.plan),
                'plan_sha256': sha256_bytes(args.plan.read_bytes()),
                'changes': len(plan['operations']),
                'unchanged': plan['unchanged_count'],
            }
        else:
            result = {
                'mode': 'apply',
                **apply_plan(
                    args.records,
                    args.plan,
                    args.output,
                    args.approve_hash,
                ),
            }
    except (WorkflowError, OSError, csv.Error) as exc:
        print(f'error: {exc}', file=sys.stderr)
        return 1

    print(json.dumps(result, ensure_ascii=False, sort_keys=True))
    return 0


if __name__ == '__main__':
    raise SystemExit(main())

実行順と入出力

1. 変更計画を作る

python workflow.py plan --records records.csv --suggestions suggestions.csv --plan plan.json

標準出力には、変更件数、変更なしの件数、計画ファイルのSHA-256ハッシュがJSONで1行出力されます。この例では変更2件、変更なし1件です。元のrecords.csvは変更されません。

計画内容は次のコマンドで読みやすく表示できます。

python -m json.tool plan.json

operationsにあるrecord_id、before、after、reasonを確認します。既存の計画を作り直す場合だけ--replace-planを指定します。置き換えると以前のハッシュは一致しなくなるため、再確認が必要です。

2. 確認した計画のハッシュを取得する

レビューを終えた後、次のコマンドで計画ファイルの現在のハッシュを表示します。

python -c "from pathlib import Path; import hashlib; print(hashlib.sha256(Path('plan.json').read_bytes()).hexdigest())"

表示された64桁の値を、確認済み計画の識別子として控えます。ハッシュを取得した後に計画ファイルを編集した場合は、内容を再確認してハッシュも取り直します。

3. 計画を適用する

次のHASHを、直前に取得した実際の値へ置き換えて実行します。

python workflow.py apply --records records.csv --plan plan.json --output updated_records.csv --approve-hash HASH

成功時は適用件数、出力先、計画ハッシュが標準出力へJSONで出ます。生成されるupdated_records.csvの論理的な内容は次のとおりです。

record_id,name,status
A001,青木商店,approved
A002,北川企画,rejected
A003,南製作所,approved

元のrecords.csvは残ります。CSVは再生成されるため、引用符、BOM、改行コードなどのバイト表現は元ファイルと異なる場合があります。

失敗時の動作

  • 利用方法の誤りはargparseが標準エラーへ案内を出し、終了コード2で終了します。
  • 入力不備、ハッシュ不一致、既存の出力先など、想定内の業務エラーは標準エラーへ理由を出し、終了コード1で停止します。
  • 全操作の検証が終わるまで出力先へ切り替えないため、途中の操作だけを反映したCSVは残りません。
  • 一時ファイルは出力先と同じディレクトリへ作り、書き込み完了後にos.replaceで出力先へ切り替えます。

たとえば計画作成後にrecords.csvを編集すると、適用時に「入力CSVが計画作成後に変化しています」と表示され、出力CSVは作成されません。改行コードやBOMだけの変更もハッシュ差として検出します。これは、論理的に同じに見える別ファイルを誤って適用しないための保守的な動作です。

最小テストを追加する

次をtest_workflow.pyとして保存します。正常適用、計画作成後の入力変更、承認後の計画変更を検査します。

import csv
import tempfile
import unittest
from pathlib import Path

from workflow import WorkflowError, apply_plan, make_plan, sha256_bytes


RECORDS = '''record_id,name,status
A001,青木商店,pending
A002,北川企画,pending
A003,南製作所,approved
'''

SUGGESTIONS = '''record_id,proposed_status,reason
A001,approved,必要項目を確認済み
A002,rejected,連絡先が不足
A003,approved,現状維持
'''


class WorkflowTest(unittest.TestCase):
    def prepare(self, root: Path) -> tuple[Path, Path, Path]:
        records = root / 'records.csv'
        suggestions = root / 'suggestions.csv'
        plan = root / 'plan.json'
        records.write_text(RECORDS, encoding='utf-8')
        suggestions.write_text(SUGGESTIONS, encoding='utf-8')
        return records, suggestions, plan

    def test_apply_changes_only_planned_rows(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            records, suggestions, plan = self.prepare(root)
            output = root / 'updated.csv'

            plan_data = make_plan(records, suggestions, plan, replace=False)
            approval = sha256_bytes(plan.read_bytes())
            result = apply_plan(records, plan, output, approval)

            self.assertEqual(len(plan_data['operations']), 2)
            self.assertEqual(result['applied'], 2)
            self.assertEqual(
                sha256_bytes(records.read_bytes()),
                plan_data['records_sha256'],
            )

            with output.open(encoding='utf-8', newline='') as file:
                rows = list(csv.DictReader(file))
            statuses = {row['record_id']: row['status'] for row in rows}
            self.assertEqual(
                statuses,
                {
                    'A001': 'approved',
                    'A002': 'rejected',
                    'A003': 'approved',
                },
            )

    def test_rejects_changed_input(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            records, suggestions, plan = self.prepare(root)
            output = root / 'updated.csv'

            make_plan(records, suggestions, plan, replace=False)
            approval = sha256_bytes(plan.read_bytes())
            changed = RECORDS.replace(
                'A001,青木商店,pending',
                'A001,青木商店,approved',
            )
            records.write_text(changed, encoding='utf-8')

            with self.assertRaisesRegex(
                WorkflowError,
                '入力CSVが計画作成後に変化',
            ):
                apply_plan(records, plan, output, approval)
            self.assertFalse(output.exists())

    def test_rejects_changed_plan(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            records, suggestions, plan = self.prepare(root)
            output = root / 'updated.csv'

            make_plan(records, suggestions, plan, replace=False)
            approval = sha256_bytes(plan.read_bytes())
            plan.write_text(
                plan.read_text(encoding='utf-8') + '\n',
                encoding='utf-8',
            )

            with self.assertRaisesRegex(
                WorkflowError,
                '計画ファイルが承認時の内容と一致しません',
            ):
                apply_plan(records, plan, output, approval)
            self.assertFalse(output.exists())


if __name__ == '__main__':
    unittest.main()

テストは次のコマンドで実行します。

python -m unittest -v test_workflow

3つのテストがokになれば、サンプルの正常系、元データ変更の拒否、計画ファイル変更の拒否を確認できます。

この実装で保証しないこと

SHA-256ハッシュはファイル内容の一致確認には使えますが、誰が承認したかを証明するものではありません。ハッシュを渡すだけで承認者の認証や電子署名になるわけでもありません。チーム運用では、計画ファイルとハッシュをアクセス制御されたレビューやCIの承認手順へ渡し、提案者、確認者、適用日時を別途記録します。

また、この例は同時実行ロック、大容量CSVのストリーミング、状態遷移ごとの権限制御を実装していません。複数プロセスが同じ出力先を扱う場合は排他制御を追加し、大容量データでは変更対象をデータベースや不変スナップショットへ移す設計を検討してください。表計算ソフトで開くCSVに外部由来の文字列を含める場合は、数式として解釈される文字列への対策も別途必要です。

導入時のチェックリスト

  • AIの出力を正本へ直接書き込まず、提案ファイルとして分離する。
  • 許可する値、必須列、重複ID、未知のIDを機械的に検証する。
  • 変更前後の値と理由を計画ファイルで確認する。
  • 確認後に取得した計画ファイルのハッシュを適用時に照合する。
  • 元データを上書きせず、出力結果を比較してから次工程へ渡す。

最初は実際の業務CSVを少量だけ複製し、許可状態と必須列を自社のルールへ置き換えて試してください。AIの判断精度とは別に、誤った提案を適用前に止められる工程を持つことが重要です。

第154回 実務で使える増分処理と冪等性の設計:Pythonで作る差分検出・重複除外・安全な書き込みパターン

はじめに — つまずきに寄り添う

増分処理や冪等性は、現場で繰り返し失敗すると面倒さが身にしみるテーマです。処理が重複してコストが増えたり、途中で止まったときに再実行してデータが壊れたりすると、日常運用が不安定になります。本記事では「現場ですぐ使える」Pythonパターンに絞り、差分検出から重複除外、冪等な書き込み、再実行・回復の設計まで段階的に解説します。シリーズ: AIとPythonの実務

なぜ増分処理と冪等性が重要か

  • コスト削減:フル処理を避け、対象の差分だけ処理することで時間・APIコール・計算資源を節約できます。
  • 再実行の安全性:途中失敗後のリトライで重複や不整合を避けられます。
  • データ一貫性:一貫した状態での保存・集計が可能になり、監査性も向上します。

増分検出の主要戦略と選び方

代表的な戦略を比較表で示します。運用コストや導入しやすさを見て選んでください。

戦略 メリット 欠点 運用コスト
mtime / ファイル更新時刻 簡単、ファイルシステム依存で高速 時刻の変更で誤検出、タイムゾーン/NFS の問題 低
シーケンスID/タイムスタンプ 順序が安定していれば増分を追いやすく、API向け ソース側のサポートが必要。同一タイムスタンプや遅延到着への対策も必要 中
ハッシュ比較(コンテンツハッシュ) 正確に差分を検出、内容ベース 計算コストが高い(大ファイル)、部分差分の検出が難しい 中〜高
差分ログ(change log / CDC) 完全な変更履歴、正確な再現性 導入コストが高い、ログ保持の管理が必要 高

Pythonでの差分検出パターン(ハンズオン指針)

1) mtimeチェック(ファイル単位)

シンプルで初期導入が容易。例:

from pathlib import Path

def changed_files(src_dir, last_run_time):
    for p in Path(src_dir).rglob('*'):
        try:
            if not p.is_file():
                continue
            mtime = p.stat().st_mtime
        except OSError:
            continue
        if mtime > last_run_time:
            yield p

注意点:rglob('*')はディレクトリも列挙するため、is_file()で通常ファイルだけに絞ります。また、タイムスタンプの精度やNFSタイムスタンプの不一致を考慮してください。

2) 行単位ハッシュ(内容ベース)

CSVやログの差分検出に有効。行ごとにハッシュを取り、既存ハッシュセットと照合します。

import hashlib

def line_hash(line: str) -> str:
    return hashlib.sha256(line.encode('utf-8')).hexdigest()

seen = set()  # 永続化するのが実運用
new_rows = []
for line in open('data.csv'):
    h = line_hash(line)
    if h not in seen:
        seen.add(h)
        new_rows.append(line)

3) ウォーターマーク管理(SQLiteを使う例)

処理位置を小さなDBで管理します。タイムスタンプだけを保存すると、同じタイムスタンプを持つ複数レコードや遅れて到着したデータを取りこぼす可能性があります。そのため、タイムスタンプと一意IDの複合カーソルを使い、(timestamp, unique_id)の順で処理位置を管理します。

import sqlite3

def init_db(path='meta.sqlite'):
    conn = sqlite3.connect(path)
    conn.execute('''
        CREATE TABLE IF NOT EXISTS watermark (
            name TEXT PRIMARY KEY,
            timestamp TEXT NOT NULL,
            unique_id TEXT NOT NULL
        )
    ''')
    conn.commit()
    return conn

def get_watermark(conn, name):
    cur = conn.execute(
        'SELECT timestamp, unique_id FROM watermark WHERE name=?',
        (name,)
    )
    row = cur.fetchone()
    return (row[0], row[1]) if row else None

def set_watermark(conn, name, timestamp, unique_id):
    conn.execute('''
        INSERT INTO watermark(name, timestamp, unique_id)
        VALUES (?, ?, ?)
        ON CONFLICT(name) DO UPDATE SET
            timestamp=excluded.timestamp,
            unique_id=excluded.unique_id
    ''', (name, timestamp, unique_id))
    conn.commit()

取得側では、日時を同一形式・同一タイムゾーンに正規化したうえで、概念的にはtimestamp > 前回timestamp OR (timestamp = 前回timestamp AND unique_id > 前回unique_id)という条件を使い、同じ順序で並べます。一意IDには、順序比較が安定する値を使用してください。

遅延到着があり得る場合は、前回位置より少し前から重複区間を再取得し、一意キーによるupsertや処理済みIDで重複を除外します。ウォーターマークは対象データの書き込みが成功した後にだけ更新し、可能であればデータ更新と同じトランザクションに含めます。

重複除外と冪等な書き込みの実装パターン

  • upsert設計:データベース側の一意キーを用いてINSERT … ON CONFLICTまたはREPLACEを利用する。
  • トランザクション的書き込み:一連の更新をトランザクションでまとめ、途中失敗時には全体をロールバック。
  • 一時ファイル→原子置換:出力はまずtempに書き込み、最終的にos.replaceで置き換える。これで中途半端なファイル状態の露出を防ぐ。
  • ファイルロック・ポジション保存:長時間処理や並列処理ではロックで同時書き込みを回避。位置情報を保存して部分再開も可能。

原子置換の最小例(atomic write):

import os
from tempfile import NamedTemporaryFile

def atomic_write(path, data_bytes):
    dirpath = os.path.dirname(path) or '.'
    with NamedTemporaryFile(dir=dirpath, delete=False) as tmp:
        tmp.write(data_bytes)
        tmp.flush()
        tmp_name = tmp.name
    os.replace(tmp_name, path)  # 原子的に置換

SQLiteでの簡単なupsert例:

# テーブル: items(id PRIMARY KEY, data TEXT)
conn.execute('INSERT INTO items(id,data) VALUES (?,?) ON CONFLICT(id) DO UPDATE SET data=excluded.data', (id, data))
conn.commit()

API連携での増分処理

  • バッチ化:API呼び出しはまとめて行い、オフセットや複合カーソルをウォーターマークで管理します。
  • 並列呼び出し時の重複リスク:位置管理に加え、APIが対応している場合はidempotency-keyを付与してAPI側で重複排除できる設計にします。
  • de-dupキーの設計:ソースの一意ID+バッチIDやコンテンツハッシュを候補にします。

idempotency-keyは試行ごとではなく、同じ論理処理に対して一度だけ生成します。送信前に処理状態とともに永続化し、成功が確定するまでは再試行でも同じキーを再利用します。以下はSQLiteにキーを保存する例です。

import sqlite3
import uuid
import requests

def init_request_db(path='meta.sqlite'):
    conn = sqlite3.connect(path)
    conn.execute('''
        CREATE TABLE IF NOT EXISTS api_requests (
            operation_id TEXT PRIMARY KEY,
            idempotency_key TEXT NOT NULL,
            status TEXT NOT NULL
        )
    ''')
    conn.commit()
    return conn

def get_or_create_key(conn, operation_id):
    row = conn.execute(
        'SELECT idempotency_key FROM api_requests WHERE operation_id=?',
        (operation_id,)
    ).fetchone()
    if row:
        return row[0]

    key = str(uuid.uuid4())
    conn.execute(
        'INSERT INTO api_requests(operation_id, idempotency_key, status) VALUES (?, ?, ?)',
        (operation_id, key, 'pending')
    )
    conn.commit()  # API送信前に保存する
    return key

def post_with_idempotency(conn, operation_id, url, payload):
    key = get_or_create_key(conn, operation_id)
    headers = {'Idempotency-Key': key}

    # 例外時もpendingのレコードとキーが残るため、
    # 同じoperation_idでの再試行では同じキーが使われる
    resp = requests.post(
        url,
        json=payload,
        headers=headers,
        timeout=30
    )
    resp.raise_for_status()

    conn.execute(
        'UPDATE api_requests SET status=? WHERE operation_id=?',
        ('succeeded', operation_id)
    )
    conn.commit()
    return resp

operation_idにはバッチIDなど、同じ論理処理を再試行しても変わらない識別子を指定します。この方法は、連携先APIがIdempotency-Keyと同一キーの再送をサポートしていることが前提です。

再実行・失敗からの回復設計

  • チェックポイントの粒度:ファイル単位、バッチ単位、レコード単位など。処理コストと回復の細かさで決める。
  • 部分再実行:複合カーソルや処理済みIDのリストで未処理部分のみ再実行する。遅延到着がある場合は重複区間を再取得する。
  • 冪等なリトライ戦略:リトライ前提なら、同じ論理処理で同じidempotencyキーを再利用するか、upsertを利用して重複を許容しない設計にする。
  • 監査ログ:処理開始時/終了時/失敗時にキー情報(ウォーターマーク、バッチID、idempotencyキー、件数、エラー)を残す。

テストとCIで検証する項目

差分・重複・削除などのケースをユニットと統合で検証します。モックを使って外部APIやファイルシステムの失敗を再現しましょう。

  • 差分ケース:追加・更新・削除・同一データの重複・同一タイムスタンプ・遅延到着を網羅。
  • モックAPI:レスポンス遅延や部分エラーを再現し、同じ論理処理のリトライでidempotencyキーが変わらないことを確認。
  • CIでの回帰確認ポイント:ウォーターマークの更新、重複率閾値、処理スループット。

運用チェックリストとモニタリング指標

項目 説明 / 計測方法 推奨閾値例
スループット レコード/秒、バッチあたり処理件数 業務要件に依存
重複率 同一IDやハッシュが処理済みで再度処理された割合 <1%(目安)
遅延時間 データ生成から処理完了までの時間 SLAに応じて設定
ウォーターマーク整合性 ウォーターマークと実処理の不整合チェック 不整合は即アラート

実務テンプレートと次の一歩

まずは小さなCLIテンプレートで試してください。argparseベースの最小構成例:

import argparse

def main():
    p = argparse.ArgumentParser()
    p.add_argument('--src', required=True)
    p.add_argument('--meta', default='meta.sqlite')
    args = p.parse_args()
    # 差分検出 → 処理 → ウォーターマーク更新 の流れを書く

if __name__ == '__main__':
    main()

配布案内:この記事で示した小さなリファレンス実装(CLIテンプレート、SQLiteメタ管理、atomic writeユーティリティ)をManage AIのリポジトリで配布予定です(公開予定日付近に案内します)。次回は「運用の自動化と監査強化」を取り上げる予定です。

まとめ

増分処理と冪等性は、現場の安定稼働に直結する重要な設計要素です。まずはmtimeや行ハッシュ、タイムスタンプと一意IDによる複合カーソルから始め、遅延到着がある場合は重複区間の再取得と重複除外を組み合わせてください。重要なのは「完璧を目指すより、失敗から安全に回復できる設計」を段階的に導入することです。

チェックリスト要約:

  • 差分戦略を選ぶ(mtime / ID / ハッシュ / CDC)
  • 複合カーソルまたは永続的なハッシュDBで状態管理し、必要に応じて重複区間を再取得する
  • 書き込みは原子的/トランザクション的に行う(upsert, atomic write)
  • APIでは同じ論理処理のidempotencyキーを保存し、成功確定まで再利用する
  • CI・テストで差分ケースを網羅し、運用監視を設定する

Manage AI(https://manageai.online)では、実務に使える小さなツール群とテンプレートを継続的に紹介します。まずはこの記事のサンプルパターンを試し、貴社の運用要件に合わせてカスタマイズしてください。

第153回 堅牢な中断・再開対応パイプライン設計:Pythonで作るチェックポイントと状態管理の実務手順

長時間処理や外部APIを伴うワークフローでは、途中で止まってしまったときに「どこから再開するか」「重複処理をどう防ぐか」で悩むことが多いはずです。本稿は、第152回(並行処理と再実行設計)を読んだ前提で、Python(標準ライブラリ+軽量外部依存)で現場に導入しやすいチェックポイント(checkpointing)と状態管理のパターン、実装例、運用上の注意点を実務目線で整理します。まずは、現場でよくあるつまずきに寄り添いながら進めます。

現場でよくある失敗パターンとチェックポイントでの解決狙い

実際の現場で見かける代表的な問題と、チェックポイント導入で期待できる効果を整理します。

よくある問題 影響 チェックポイントでの解決策
処理途中で失敗し、全件を再実行する 時間とコストの浪費、APIレート超過 進捗を保存し、最後に成功した境界から再開
重複処理(重複課金・二重登録) 外部サービスでのコストやデータ不整合 冪等キー・トランザクション境界でスキップ判定
部分成功データの中途半端な出力 解析や後工程でのエラーの温床 原子的コミットや一時ファイル+renameで整合性を担保

基本設計パターン(実務の観点)

チェックポイント設計で押さえておくべき基本パターンを短く整理します。

  • 冪等性 (idempotency):同じ操作を複数回実行しても結果が変わらないよう、外部APIには冪等キーを渡したり、ローカルで既処理レコードを参照してスキップする。
  • トランザクション境界:「ここまで成功したら永続化する」という明確な境界を決める。境界は小さくするほど回復が細かくできるが管理コストが増える。
  • 原子的コミット:ローカルファイルは tmp → atomic rename(os.replace)で最終化する。S3ではrename相当の操作がないため、条件付き書き込みや実行ごとに一意なキーを使い、同じ最終キーへの並行書き込みを制御する。
  • 部分出力の扱い:中間出力は明示的に状態として残し、後工程はそのスキーマを前提に再実行できるようにする。
  • ロールフォワードとロールバック:失敗時はロールバックよりロールフォワード(欠損部分を補う)が現場では現実的な場合が多い。重要データはトランザクションで保護。

実装パターンと具体例(利点・注意点付き)

ここでは現場で採用しやすい4つのパターンを示します。コードはそのまま貼れるPythonスニペットです。

1) ファイルベース(tmpファイル → atomic rename)

長時間処理の中間結果をローカルファイルに書くときの基本パターン。小規模バッチに最適。

利点 注意点
依存が少なく、実装が簡単 分散環境やクラッシュでの復旧ロジックを追加する必要あり
原子的操作(os.replace)で整合性確保 ファイル破損やディスク容量に注意
import os
import json

def atomic_write(path, data):
    tmp = path + ".tmp"
    with open(tmp, 'w', encoding='utf-8') as f:
        json.dump(data, f)
        f.flush()
        os.fsync(f.fileno())
    os.replace(tmp, path)  # atomic on most OS

2) JSONL(追記ログ)

処理ごとに追記していくログ形式。イベントソーシング的に使える。

利点 注意点
追記のみなので衝突が少ない サイズ管理とGCが必要(古いログの圧縮)
追記順で再生(ロールフォワード)可能 途中での不整合行は復旧が難しい場合あり
import json
import os
from datetime import datetime

def append_event(path, event):
    line = json.dumps({"ts": datetime.utcnow().isoformat(), **event}, ensure_ascii=False)
    with open(path, 'a', encoding='utf-8') as f:
        f.write(line + "\n")
        f.flush()
        os.fsync(f.fileno())

3) SQLite(軽量状態DB)

状態管理(チェックポイント、未処理リスト、成功フラグ等)にはSQLiteが便利。トランザクションで整合性を保てます。

利点 注意点
ACID対応で安全 高並列書き込みは苦手(適切な排他が必要)
ファイルで持てるのでバックアップ容易 スキーマ変更時のマイグレーションが必要
import sqlite3

def init_db(path):
    con = sqlite3.connect(path, isolation_level=None)
    cur = con.cursor()
    cur.execute('''
    CREATE TABLE IF NOT EXISTS checkpoints (
        key TEXT PRIMARY KEY,
        state JSON,
        updated_at TEXT
    )
    ''')
    return con

def save_checkpoint(con, key, state_json):
    cur = con.cursor()
    cur.execute('REPLACE INTO checkpoints (key, state, updated_at) VALUES (?, ?, CURRENT_TIMESTAMP)', (key, state_json))

4) クラウドストレージ(S3)でのアップロード

S3ではrename操作がないため、複数の実行が固定の一時キーを共有する方式は避けます。最終キーがまだ存在しない場合だけ公開する用途では、条件付きPUTを使うことで、同じキーに対する並行実行の上書きを防げます。上書きが必要なワークフローでは、実行ごとに一意な最終キーを使うか、別途排他制御を設けてください。

利点 注意点
耐久性が高く、共有が容易 rename操作がないため、条件付き書き込みや一意なキーの設計が必要
分散環境で扱いやすい リージョンや認証まわりの運用設定に注意
# boto3 を使い、最終キーが未作成の場合だけアップロードする例
import boto3
from botocore.exceptions import ClientError

s3 = boto3.client('s3')

def upload_if_absent(bucket, key, data_bytes):
    try:
        s3.put_object(
            Bucket=bucket,
            Key=key,
            Body=data_bytes,
            IfNoneMatch='*'
        )
    except ClientError as exc:
        status = exc.response.get('ResponseMetadata', {}).get('HTTPStatusCode')
        if status == 412:
            return False  # すでに同じ最終キーが存在する
        raise
    return True

再開(復旧)ロジックの実装ポイント

処理の再実行判定やスキップロジックはシンプルに保つことが重要です。代表的なパターンを示します。

  • チェックサム(入力のハッシュ)を保存しておき、入力が変わっていなければスキップできる。
  • ステータスマーク(pending / running / success / failed)を保存し、runningが長時間続く場合は調査フラグにする。
  • 部分再処理は「最小の再実行単位(バケット)」を設計しておき、そこ単位で再実行する。

ただし、チェックポイントだけではexactly-once(厳密に1回だけの実行)は保証できません。外部処理の成功後、成功チェックポイントの保存前にプロセスが停止すると、再開時に外部処理が再実行されます。二重登録や二重課金を防ぐには、再試行しても変わらない冪等キーを外部処理へ渡し、外部サービス側でも同じキーの重複実行を抑止する必要があります。

以下の擬似例では、process()を、受け取った冪等キーを外部APIへ転送するアプリケーション側の関数として扱います。同じ項目と同じ入力には、再開後も同じ冪等キーを使用します。

import hashlib

def checksum_of_bytes(b):
    return hashlib.sha256(b).hexdigest()

# 再開ロジック(擬似)
item_key = 'item-123'
input_checksum = checksum_of_bytes(input_bytes)
idempotency_key = f'{item_key}:{input_checksum}'

state = load_checkpoint(item_key)  # state は dict
if state and state.get('status') == 'success' and state.get('input_checksum') == input_checksum:
    print('skip: already processed')
else:
    try:
        mark_running(item_key)
        result = process(
            input_bytes,
            idempotency_key=idempotency_key
        )
        save_checkpoint(item_key, {
            'status': 'success',
            'input_checksum': input_checksum
        })
    except Exception:
        save_checkpoint(item_key, {'status': 'failed'})
        raise

テストと検証

チェックポイント機能はテスト設計が重要です。単体テスト、統合テスト、障害注入テストのチェックリストを示します。

レイヤ 検証項目 方法
単体テスト ファイル書き込みが原子的に見えるか/SQLiteのトランザクション 一時ディレクトリやin-memory SQLiteで検証、mockでIOを代替
統合テスト 実際に中断→再開して期待通り結果が得られるか CIで一時S3バケットやテストDBを使って流し込み
障害注入 書き込み時のクラッシュ、ネットワーク断を再現 プロセス強制終了、モックで例外を発生させる

CIに組み込むべき検証ポイント例:

  • チェックポイントを書いた後の再起動で処理がスキップされること
  • 中間ファイルが残らない(.tmp等の掃除)こと
  • 外部APIに対する冪等キーによる二重請求防止の確認

運用と監視の実務チェックリスト

導入後に確認・監視すべきポイントを表で整理します。

監視項目 閾値/ルール 対応アクション
未完了ジョブ数 常時0に近いこと。上昇トレンドは問題 原因調査・自動再実行の検討
最終成功時刻 想定頻度を超える遅延がある場合アラート 処理負荷や外部APIの障害確認
再実行頻度 一定以上なら根本原因の調査が必要 コードや外部依存の堅牢化

ゴミデータ(古いチェックポイントや一時ファイル)は定期クリーンアップをスケジュールし、最小保持ポリシーを決めて自動化してください。

移行・互換性とメンテナンス

チェックポイントデータのスキーマ進化は実務で必ず起きます。対処方針の例:

  • checkpoint に version フィールドを追加し、読み込み時に古いバージョンを検出してマイグレーション関数を呼ぶ。
  • 重大変更時はバックフィル計画を作り、テスト環境でまず適用する。
  • 古いチェックポイントは段階的に削除するポリシーを定める(例:90日経過でアーカイブ→削除)。
# 簡単なバージョニング読み込み例
def load_state_with_migration(raw):
    version = raw.get('version', 1)
    if version == 1:
        return migrate_v1_to_v2(raw)
    return raw

まとめと次の一歩

本稿では、チェックポイント設計の目的とトレードオフ、冪等性や原子的コミットなどの基本パターン、ファイル/JSONL/SQLite/S3それぞれの実装例、再開ロジック、テスト・運用・移行のポイントを実務的に整理しました。まずは小さなバッチで「tmp→atomic rename」や「SQLiteでの状態管理」から導入し、問題が起きたら障害注入テストで洗い出すのが現場で現実的な進め方です。

テンプレートコード(抜粋)を下に示します。WordPressにそのまま貼れる形式にしてあるので、プロジェクトにコピーして試してください。

# チェックポイント簡易テンプレート
import os, json, hashlib

def atomic_write(path, data):
    tmp = path + '.tmp'
    with open(tmp, 'w', encoding='utf-8') as f:
        json.dump(data, f, ensure_ascii=False)
        f.flush(); os.fsync(f.fileno())
    os.replace(tmp, path)

def checksum(b):
    return hashlib.sha256(b).hexdigest()

def process_item(item_bytes, state_path):
    try:
        with open(state_path, 'r', encoding='utf-8') as f:
            state = json.load(f)
    except FileNotFoundError:
        state = {}

    cs = checksum(item_bytes)
    if state.get('input_checksum') == cs and state.get('status') == 'success':
        return 'skipped'

    # ... 実処理ここから
    # 外部APIを呼ぶ場合は、csなどから作った同一の冪等キーを再試行時にも渡す
    # 処理が成功したらチェックポイントを原子的に保存
    new_state = {'status': 'success', 'input_checksum': cs}
    atomic_write(state_path, new_state)
    return 'done'

次回候補としては、作ったチェックポイントをAirflowやPrefectのようなオーケストレーションツールとどう連携するか、監査ログやガバナンスとどうつなぐかを扱う予定です。本シリーズ「AIとPythonの実務」の流れで、現場で使える実装を続けて紹介します。

抜粋:長時間処理や部分失敗が発生する現場向けに、処理の中断・再開(checkpointing)と状態管理をPythonで実装する手順を解説します。第152回(並行処理と再実行設計)の続きとして、再実行可能・冪等・安全な状態保存の実務的パターンに焦点を当てます。

第152回 実務で使える並行処理と再実行設計 — Pythonで作る安全で効率的なバッチ&API呼び出しワークフロー

はじめに — 現場でよくあるつまずきに寄り添う

大量のCSVを処理したり、埋め込みや推論APIを大量に呼び出したりすると、処理が遅い・途中で失敗してやり直しが大変・コストが膨らむ、という問題に直面します。本記事は「どの並列化を選ぶか」「失敗時にどう回復するか」を、Pythonコードと実務向けのチェックリストで示します。第151回(logging/CLI/config)との連携を想定し、運用につながる形で説明します。

1) いつ並列化すべきか(事前計測の手順)

まずは並列化する前にボトルネックを把握します。小さなサンプルで計測し、I/O待ちかCPU負荷か、多接続が原因かを見極めます。

  • 測定項目:1件あたりの平均処理時間、外部APIの待ち時間、CPU使用率、メモリ増加
  • 測定方法の例:代表的な100件でwall-clock時間を測る。外部通信時間が総時間の大部分ならI/Oバウンド。
状況 並列化の候補 注目点
I/Oバウンド(ネットワーク/API) ThreadPoolExecutor / asyncio 同時接続数の制御、タイムアウト設定
CPUバウンド(大きな変換/圧縮) ProcessPoolExecutor GIL回避、pickle可能性
多数の同時接続(高並列API呼び出し) asyncio + aiohttp 等 バックプレッシャー、レート制限対応

2) パターン紹介と短いコード例

ここでは代表的な3パターンの最小構成を示します。実務では利用するAPI、ライブラリ、ネットワーク条件に合わせて、タイムアウトや例外処理を追加してください。

ThreadPoolExecutor(簡単に並列化)

外部APIを呼ぶI/Oバウンド処理で使いやすい方法です。次の例は標準ライブラリで複数URLを並行取得します。スレッド数を増やしすぎず、接続先の制限に合わせて調整してください。

from concurrent.futures import ThreadPoolExecutor
from urllib.request import urlopen


def fetch_one(url):
    with urlopen(url, timeout=10) as response:
        return response.read()


urls = [
    "https://example.com/",
    "https://example.com/",
]

with ThreadPoolExecutor(max_workers=4) as executor:
    bodies = list(executor.map(fetch_one, urls))

executor.map()は入力順に結果を返します。個別の完了順で処理したい場合や、失敗した項目を特定したい場合は、後述の実践例のようにsubmit()とwait()を使います。

ProcessPoolExecutor(CPUバウンド向け)

CPU負荷の高い処理でGILの影響を避けたい場合に使います。ワーカーへ渡す関数や引数、戻り値はpickle可能である必要があります。ワーカー関数はモジュール直下に定義します。

from concurrent.futures import ProcessPoolExecutor


def cpu_task(value):
    return sum(i * i for i in range(value))


if __name__ == "__main__":
    values = [100_000, 120_000, 140_000]
    with ProcessPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(cpu_task, values))
    print(results)

asyncio + aiohttp(高並列のAPI呼び出し)

多数の短時間リクエストを扱う場合に有効です。次の例では、1つのClientSessionを再利用し、asyncio.Semaphoreで同時実行数を制限します。aiohttpは標準ライブラリではないため、利用環境へ別途導入する必要があります。

import asyncio

import aiohttp


async def fetch_json(session, url, semaphore):
    async with semaphore:
        timeout = aiohttp.ClientTimeout(total=10)
        async with session.get(url, timeout=timeout) as response:
            response.raise_for_status()
            return await response.json()


async def main(urls):
    semaphore = asyncio.Semaphore(10)
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_json(session, url, semaphore) for url in urls]
        return await asyncio.gather(*tasks)


# results = asyncio.run(main(urls))

asyncio.gather()は、既定ではいずれかのタスクが例外を送出すると、その例外を呼び出し元へ伝えます。項目ごとに失敗を記録する場合は、各タスク内で対象となる例外を処理する設計にします。

3) レート制御・同時接続制限

APIにはレート制限やコストがあるため、同時接続数や時間当たりの呼び出し数を制御します。同時実行数だけを抑えるならSemaphore、時間当たりの許容量まで管理するならトークンバケットなどを検討します。

方法 説明 利点/注意点
Semaphore 同時実行数を単純に制限する 実装が簡単。時間当たりのリクエスト数は直接制御しない。
トークンバケット 時間当たりの許容量を制御する レート制限に適合させやすいが、実装と状態管理がやや複雑。

4) 再試行戦略とbackoff実装

一時的なネットワーク障害に対しては、回数に上限を設けた再試行を使います。すべての例外を一律に再試行するのではなく、失敗理由とAPI仕様に応じて再試行可否を判断してください。

  • 再試行を検討する例:一時的なタイムアウト、接続断、502/503など。
  • 原則として再試行しない例:認証エラー(401)、入力不備(400など)、権限不足。
  • 429の扱い:一時的なレート制限ならRetry-AfterやAPI仕様に従って待機し、上限回数内で再試行する。契約上のクォータ枯渇など、待機しても解消しない429は再試行を打ち切る。

次のコードは、再試行可能と判定済みの失敗だけをRetryableErrorとして受け取り、exponential backoffにjitterを加えて待機します。max_attemptsは初回実行を含む総試行回数です。

import random
import time


class RetryableError(Exception):
    def __init__(self, message, retry_after=None):
        super().__init__(message)
        self.retry_after = retry_after


def backoff_delay(attempt, base=1.0, cap=30.0):
    upper = min(cap, base * (2 ** attempt))
    return random.uniform(0, upper)


def call_with_retry(operation, max_attempts=5):
    for attempt in range(max_attempts):
        try:
            return operation()
        except RetryableError as exc:
            if attempt == max_attempts - 1:
                raise

            if exc.retry_after is not None:
                delay = max(0.0, exc.retry_after)
            else:
                delay = backoff_delay(attempt)

            time.sleep(delay)

実際のAPIアダプター側では、タイムアウトや一時的な502/503、再試行可能な429だけをRetryableErrorへ変換します。恒久的なエラーは変換せず、そのまま呼び出し元へ返します。429のRetry-Afterの形式や単位は利用するAPIの仕様に従って解釈してください。

外部ライブラリを利用できる環境では、tenacityなどで停止条件、待機条件、再試行対象の例外を設定する方法もあります。導入時は利用中のバージョンの公式ドキュメントに合わせて設定してください。

5) 再実行に強い設計:チェックポイント・idempotency・トランザクション的コミット

長時間バッチは途中で中断されることを想定します。チェックポイントと冪等性(idempotency)を組み合わせ、中断後に同じ処理が一部重複しても結果が壊れないようにします。

項目 実務上の設計例
入力の同一性確認 入力ファイルのハッシュ、サイズ、更新日時などを保存し、再開時に同じ入力か確認する。ハッシュだけでは処理済み位置は分からない。
進捗チェックポイント 処理済みの行ID、確定済みレコードID、または安全に再開できるオフセットを保存する。
idempotency APIやDBが対応している場合は、行IDなどの安定した一意キーを使い、重複実行でも二重登録しない仕様にする。
コミット 一時ファイルからの置換、DBトランザクション、確定状態の記録などにより、不完全な出力を完成済みとして扱わない。

並行処理では完了順が入力順と一致しないため、単純に「最後に完了した行番号」だけを保存すると、未完了行を飛ばす可能性があります。処理済みIDを保存するか、順序どおりに確定した連続位置だけをチェックポイントとして記録してください。

6) 優雅な終了と再開(SIGTERM対応・中断保存)

実行中に終了要求を受けた場合は、新しい処理の投入を止め、実行中の処理を可能な範囲で完了させてからチェックポイントを保存します。Unix系環境ではSIGTERMを捕捉する方法が一般的です。

import signal
import threading


stop_requested = threading.Event()


def request_stop(signum, frame):
    stop_requested.set()


if hasattr(signal, "SIGTERM"):
    signal.signal(signal.SIGTERM, request_stop)
signal.signal(signal.SIGINT, request_stop)

シグナルハンドラー内では複雑なI/Oを行わず、終了要求のフラグだけを設定します。メイン処理がそのフラグを確認し、安全な位置でチェックポイントを書き出します。保存頻度はコストとのトレードオフであり、数秒ごと、N件ごと、または成功ごとに保存する方法があります。

7) 監視とログ(運用につなげる)

ログは詳細すぎても不足でも困ります。エラー原因の特定に必要な情報として、リクエストID、行ID、エラーコード、再試行回数、最終的な成否を記録します。認証情報や入力本文などの機密情報をそのまま出力しないよう注意してください。第151回の設定と組み合わせ、CLIで並列度やタイムアウトを外から調整できるようにしておくと便利です。

8) 実践例:CSV→埋め込みAPI→ベクトルDB(最小構成)

ここでは、CSVを逐次読み込み、ThreadPoolExecutorで処理し、成功したIDをチェックポイントへ保存する最小構成を示します。特定事業者のAPIやベクトルDBには依存せず、process_rowへ「API呼び出しと保存」を行う関数を渡す形です。

ステップ ポイント
1. ストリーミング読込 CSVを1行ずつ読み、全件を一度にメモリへ載せない。
2. 同時実行数の制限 保留中のFuture数に上限を設け、読み込み側へバックプレッシャーをかける。
3. API呼び出し タイムアウトを設定し、再試行可能な失敗だけをRetryableErrorへ変換する。
4. 成功時処理 ベクトルDBなどへの保存が成功した後に、処理済みIDをチェックポイントへ追加する。
5. 失敗時処理 最大試行回数を超えた項目をエラーファイルへ記録する。
6. 終了/再開 入力ハッシュを確認し、処理済みIDを除外して再開する。

以下のコードは、前節のcall_with_retryとstop_requestedが定義済みであることを前提にしています。CSVには重複しないid列が必要です。

import csv
import hashlib
import json
import os
from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait


def file_sha256(path):
    digest = hashlib.sha256()
    with open(path, "rb") as source:
        for chunk in iter(lambda: source.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def load_checkpoint(path, input_sha256):
    if not os.path.exists(path):
        return set()

    with open(path, "r", encoding="utf-8") as source:
        data = json.load(source)

    if data["input_sha256"] != input_sha256:
        raise ValueError("入力ファイルがチェックポイント作成時と異なります")

    return set(data.get("completed_ids", []))


def save_checkpoint(path, input_sha256, completed_ids):
    temporary_path = path + ".tmp"
    data = {
        "input_sha256": input_sha256,
        "completed_ids": sorted(completed_ids),
    }

    with open(temporary_path, "w", encoding="utf-8") as destination:
        json.dump(data, destination, ensure_ascii=False)
        destination.flush()
        os.fsync(destination.fileno())

    os.replace(temporary_path, path)


def append_error(path, row_id, exc):
    record = {
        "id": row_id,
        "error_type": type(exc).__name__,
        "message": str(exc),
    }
    with open(path, "a", encoding="utf-8") as destination:
        destination.write(json.dumps(record, ensure_ascii=False) + "\n")


def run_csv(input_path, process_row, max_workers=8, max_attempts=5):
    checkpoint_path = input_path + ".checkpoint.json"
    error_path = input_path + ".errors.jsonl"
    input_sha256 = file_sha256(input_path)
    completed_ids = load_checkpoint(checkpoint_path, input_sha256)
    max_pending = max_workers * 2

    def submit_row(executor, row):
        return executor.submit(
            call_with_retry,
            lambda: process_row(row),
            max_attempts,
        )

    def collect_finished(pending, block_until_one=False):
        if not pending:
            return

        timeout = None if block_until_one else 0
        done, _ = wait(
            pending,
            timeout=timeout,
            return_when=FIRST_COMPLETED,
        )

        for future in done:
            row_id = pending.pop(future)
            try:
                future.result()
            except Exception as exc:
                append_error(error_path, row_id, exc)
            else:
                completed_ids.add(row_id)
                save_checkpoint(
                    checkpoint_path,
                    input_sha256,
                    completed_ids,
                )

    pending = {}
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        with open(input_path, newline="", encoding="utf-8") as source:
            reader = csv.DictReader(source)

            if not reader.fieldnames or "id" not in reader.fieldnames:
                raise ValueError("CSVにはid列が必要です")

            for row in reader:
                if stop_requested.is_set():
                    break

                row_id = row["id"]
                if not row_id:
                    raise ValueError("idが空の行があります")
                if row_id in completed_ids:
                    continue

                while len(pending) >= max_pending:
                    collect_finished(pending, block_until_one=True)

                future = submit_row(executor, row)
                pending[future] = row_id
                collect_finished(pending)

        while pending:
            collect_finished(pending, block_until_one=True)

process_row(row)は、対象APIを呼び出して埋め込みを取得し、ベクトルDBなどへの保存が完了した時点で正常終了する関数として実装します。一時障害はRetryableErrorとして送出し、認証エラーや入力不備などは再試行対象にしません。

APIまたは保存先がidempotency keyや一意制約を提供している場合は、CSVのidを安定したキーとして利用します。保存成功直後、チェックポイント更新前にプロセスが強制終了すると、再開時に同じ行が再実行される可能性があるためです。外部APIが冪等性を保証しない場合は、事前照会、DBの一意制約、処理状態テーブルなど、保存先に合った重複防止策が必要です。

この最小構成は処理済みIDをJSONへ保存するため、件数が非常に多い場合はチェックポイントファイルも大きくなります。その場合は、DBの状態テーブルや確定済みの連続オフセットなどへ置き換えてください。

9) 注意点と落とし穴

  • GILの存在:CPUバウンドはProcessPoolを検討する。
  • 外部APIのレート制限とコスト:過度な並列は高額請求やAPIブロックを招く。
  • pickleできないオブジェクト:ProcessPoolへ渡す関数、引数、戻り値に注意する。
  • 再試行による重複:書き込みを伴う処理ではidempotencyや一意制約を設計する。
  • チェックポイントの誤用:入力ハッシュは同一性確認用であり、進捗情報の代わりにはならない。
  • ログの過不足:問題追跡に必要な情報を確保しつつ、機密情報を記録しない。

まとめ — 実務で取り入れるための短い手順

まずは小さなサンプルでボトルネックを計測し、I/O、CPU、多接続のどれが原因かを見極めます。I/OバウンドならThreadPoolかasyncio、CPUバウンドならProcessPoolを選択します。再試行は対象となる失敗を限定し、exponential backoffとjitter、APIが指定する待機時間を組み合わせます。入力の同一性確認と進捗チェックポイントを分け、idempotencyによって途中中断から安全に再開できるようにします。運用設定(並列度・タイムアウト・再試行回数)はCLIやconfigで外部化し、監視とログで挙動を把握してください。

読後のアクションリスト(短期)

  • 現行バッチのボトルネック計測を行う
  • 小さなサンプルでThreadPoolまたはasyncioを試す
  • 再試行対象と再試行しないエラーを明文化する
  • 入力の同一性情報と進捗チェックポイントを分けて保存する
  • 本番前に低負荷で運用テストを行い、ログと再開動作を確認する

次回は、実際のCLIテンプレートと組み合わせた簡易フレームワークを提示し、Cronやスケジューラとの安全な連携方法を紹介します。シリーズ「AIとPythonの実務」の流れで、実務で使える小さな改善を積み重ねていきましょう。

参考リンク:Manage AI 第151回(logging/CLI/config)の設定を先に確認すると運用へスムーズにつなげられます。