PythonとAIでCSV集計を自動化する前に｜検算を組み込む

更新日：2026年11月13日

CSV集計をAIへ依頼する前に、金額の単位、欠損・重複の扱い、検算の正解を決めます。この記事では10行のダミーデータを部門別に集計し、異常4行を理由付きで残すところまで作ります。

対象はターミナルでPythonを実行できる開発者・業務担当者です。掲載コードは編集時に作成・実行した教材で、AIサービスへ実データを送った事例ではありません。入力から出力までローカルだけで動作します。

集計仕様を先に決める

この例は丸めを必要としません。円未満や返品のマイナス金額を扱う仕事なら仕様を変える必要があります。浮動小数点で金額を計算する場合の誤差を避けたいときは、PythonのDecimalと丸め規則を検討します。

CSVは行をカンマでsplitする代わりに csv.DictReader で読みます。引用符付きの値などを解析でき、値は基本的に文字列で返るので、この教材では型を明示して検証します。ファイルを newline="" で開く扱いもPython csv公式ドキュメントに従います。

1. 全10行を保存し、手計算する

検証環境はmacOS（Darwin arm64）、Python 3.14.2です。外部ライブラリ、APIキー、ネット接続は不要です。Pythonが入った端末で python3 --version を確認し、空の作業フォルダーを作ってください。Windowsでは以下の python3 を、利用環境に応じて py に読み替えます。掲載コード・入力はすべて架空の学習用です。

以下をUTF-8の sample.csv として保存します。

id,date,amount,department
1,2026-09-01,1000,A
2,2026-09-02,2000,A
3,2026-09-03,500,B
4,2026-09-04,0,B
5,2026-09-05,250,A
6,2026-09-06,750,B
7,2026-09-07,,A
2,2026-09-08,999,A
9,2026-02-30,100,B
10,2026-09-10,-50,A

正常なidは1〜6です。Aは1000 + 2000 + 250 = 3250円、Bは500 + 0 + 750 = 1250円、合計4500円になります。

行番号には1行目のヘッダーを含みます。空の物理行はcsvモジュールが読み飛ばすため、以下のカウンターは一般的なファイル行数ではなく、読み取ったデータレコード数です。掲載データには空行・セル内改行を含めていません。セル内改行がある入力では、記録のlineはレコード順の番号として扱ってください。

2. 完成コードで集計する

以下を csv_summary.py に保存します。

import csv
import json
import re
import sys
from collections import defaultdict
from datetime import date
from pathlib import Path

FIELDS = ["id", "date", "amount", "department"]

def summarize(path):
    totals = defaultdict(int)
    errors, seen = [], set()
    count = accepted = 0
    with Path(path).open(encoding="utf-8-sig", newline="") as handle:
        reader = csv.DictReader(handle)
        if reader.fieldnames != FIELDS:
            raise ValueError("header must be id,date,amount,department")
        for line, row in enumerate(reader, 2):
            count += 1
            try:
                if None in row or any(v is None or not v.strip() for v in row.values()):
                    raise ValueError("missing or extra field")
                row = {k: v.strip() for k, v in row.items()}
                if row["id"] in seen:
                    raise ValueError("duplicate id")
                if not re.fullmatch(r"[0-9]{4}-[0-9]{2}-[0-9]{2}", row["date"]):
                    raise ValueError("date format")
                date.fromisoformat(row["date"])
                if not re.fullmatch(r"[0-9]+", row["amount"]):
                    raise ValueError("amount must be nonnegative integer yen")
                amount = int(row["amount"])
                if row["department"] not in {"A", "B"}:
                    raise ValueError("unknown department")
                seen.add(row["id"])
                totals[row["department"]] += amount
                accepted += 1
            except ValueError as exc:
                errors.append({"line": line, "row": row, "reason": str(exc)})
    assert count == accepted + len(errors)
    return {"rows": count, "accepted": accepted, "rejected": len(errors),
            "totals": dict(sorted(totals.items())), "errors": errors}

if __name__ == "__main__":
    source = Path(sys.argv[1])
    output = Path(sys.argv[2])
    if source.resolve() == output.resolve() or output.exists():
        raise SystemExit("output must be a new file distinct from input")
    result = summarize(source)
    with output.open("x", encoding="utf-8") as handle:
        json.dump(result, handle, ensure_ascii=False, indent=2)
    print(json.dumps({k: v for k, v in result.items() if k != "errors"}, ensure_ascii=False))

python3 csv_summary.py sample.csv result.json

2026年10月5日に実行した標準出力です。

{"rows": 10, "accepted": 6, "rejected": 4, "totals": {"A": 3250, "B": 1250}}

result.json にはこの集計値に加えて、4件のerrorsが保存されます。rows = accepted + rejectedの一致をコード内で検査しています。入力のsample.csvには書き込みません。同じコマンドを再実行すると出力がすでにあるため停止します。比較を残したい場合は result-2.json など別名を指定してください。

3. 正常・異常を5テストで検算する

以下を同じフォルダーの test_csv_summary.py に保存します。期待値は集計処理の出力からコピーせず、先ほどの手計算と仕様から決めています。

import tempfile
import unittest
from pathlib import Path
from csv_summary import summarize

class CsvChecks(unittest.TestCase):
    def run_csv(self, body):
        with tempfile.TemporaryDirectory() as folder:
            path = Path(folder) / "input.csv"
            path.write_text("id,date,amount,department\n" + body, encoding="utf-8")
            return summarize(path)

    def test_sample(self):
        result = summarize(Path(__file__).with_name("sample.csv"))
        self.assertEqual((result["rows"], result["accepted"], result["rejected"]), (10, 6, 4))
        self.assertEqual(result["totals"], {"A": 3250, "B": 1250})
        self.assertEqual([r["line"] for r in result["errors"]], [8, 9, 10, 11])

    def test_empty(self):
        self.assertEqual(self.run_csv("")["accepted"], 0)

    def test_zero(self):
        self.assertEqual(self.run_csv("1,2026-09-01,0,A\n")["totals"], {"A": 0})

    def test_bad_amounts(self):
        for value in ["", "1.5", "NaN", "-1", "abc"]:
            with self.subTest(value=value):
                self.assertEqual(self.run_csv(f"1,2026-09-01,{value},A\n")["rejected"], 1)

    def test_duplicate(self):
        result = self.run_csv("1,2026-09-01,100,A\n1,2026-09-02,200,A\n")
        self.assertEqual(result["totals"], {"A": 100})
        self.assertEqual(result["rejected"], 1)

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

python3 -m unittest -v test_csv_summary

実行結果は Ran 5 tests / OK、終了コード0でした。テスト対象は10行の集計、ヘッダーだけの空データ、0円、欠損・小数・NaN・負数・文字列、重複idです。小数などは5つのsubTestで検証するため、表示されるテストメソッド数は5です。

入力を出力先にも指定する失敗例も確認できます。

python3 csv_summary.py sample.csv sample.csv

output must be a new file distinct from input

この操作は終了コード1で停止しました。異常行を見つけた通常の集計は終了コード0で結果を保存する仕様です。終了コードだけで業務利用可能と判断せず、errorsを確認して原データを直してください。列名違いなど、処理全体が成立しない場合はValueErrorで停止します。

結果が合わないときは、理由と元データを照合する

この教材の正常結果は「10行・採用6行・除外4行・合計4500円」です。合計だけが一致しても、別の行を除外して相殺している可能性があるため、採用件数と除外理由も照合します。

例えば、id=2の999円が訂正後の金額だった場合、この教材は「後の値で更新」せず除外します。最初の正常行を採る仕様だからです。業務上は最新の取引を採りたいなら、更新日時・取消の扱い・訂正前後の検算を決めてからコードとテストを変更してください。これは架空の判断例であり、実データの処理結果ではありません。

「除外4件」を確認せず4500円だけを別システムに渡すと、未処理の取引を落とすおそれがあります。担当者が除外理由を確認し、修正後の入力を別名で保存して再集計するところまでを作業に含めます。

AIへコード作成・変更を依頼する文面

公開できるダミーCSVと上記の仕様表を渡し、次のように依頼します。実データは社内の権限・保存先・情報管理を確認してから扱ってください。

Python標準ライブラリで、id,date,amount,departmentのCSVを部門別に集計してください。0以上の整数円だけ採用し、欠損・無効日付・未知部門・採用済みidの重複を理由付きで出力してください。入力・既存出力を上書きせず、行数の一致とA3250円、B1250円を検算するunittestも作成してください。仕様を変える提案はコード変更と分けて説明してください。

AIの回答を採用するときは、まずこの10行と同じテストを実行します。処理の簡略化でerrorsが消えた、負数を黙って採用した、重複を勝手に合算した場合は仕様と照合します。

実データへ広げる前の判断

この教材で扱っていないのは、通貨混在、同一idの更新履歴、返金、セル内改行を含む物理行番号、大容量ファイル、部署マスターの変更です。実際の運用ではidが一意な取引IDかを最初に確認してください。同じidの行が正当な明細なら、今回の重複ルールは使えません。

読者の完了条件は、結果が手計算と一致し、4異常行を理由付きで説明でき、テストと上書き防止の停止を再現できることです。自分の業務データに置き換えるときは、担当者が正解を知っているサンプルから検算を始めます。

AI業務改善の選び方では、内製かツール利用かの判断材料を整理しています。

出典と確認日

• Python csv、unittest、decimal（2026年10月5日確認）
• ローカル検証：2026年10月5日、Python 3.14.2 / Darwin arm64。外部AIによる生成・実データでの運用は未実施。
• 2026年11月13日：掲載コードを再抽出し、5テスト、A3250円・B1250円の検算、入力への上書き拒否を再確認。エラー別の確認表と訂正データの判断例を追加。