Python講座 / 実習：CSVを検証して集計レポートを作る
CSVの集計では、見出し、必須項目、金額と重複IDを検証してから合計します。拒否した行と失敗の終了コードを報告すると、一部のデータだけのレポートを完全な結果と取り違えずに済みます。
入力と拒否の規則を決める
idは支出を識別し、categoryは分類、amountは十進数字で書いた非負の整数です。最初に受理した有効なレコードがIDを確保し、後の同一IDを拒否します。入力ファイルを直さず、拒否理由を記録します。
新しいフォルダーを使います。report.csvかrejected.csvが既にあれば停止します。記録番号は見出しを1とする論理CSVレコード番号で、引用符内に改行があると実際のファイル行数とは異なります。
検証用の入力全体を保存する
expenses-input.csvへ保存します。重複IDと負の金額は、拒否経路を試すための意図的な不正データです。
id,category,amount e1,food,3500 e2,food,8000 e3,transport,2500 e2,food,8000 e4,books,-1
不正な値が説明なしに消えることを成功と見なさず、拒否理由と件数へ反映されるかを確認します。
レポートプログラムを実行する
csv_report.pyへ保存します。
import csv
import re
from collections import defaultdict
from pathlib import Path
source = Path("expenses-input.csv")
report = Path("report.csv")
rejected = Path("rejected.csv")
if report.exists() or rejected.exists():
raise FileExistsError("Use a new output directory")
totals = defaultdict(int)
seen = set()
errors = []
accepted = 0
with source.open(encoding="utf-8-sig", newline="") as f:
reader = csv.DictReader(f)
if reader.fieldnames != ["id", "category", "amount"]:
raise ValueError("unexpected header")
for number, row in enumerate(reader, start=2):
key = (row.get("id") or "").strip()
category = (row.get("category") or "").strip()
amount_text = (row.get("amount") or "").strip()
reason = ""
if None in row or not key or not category:
reason = "invalid fields"
elif key in seen:
reason = "duplicate id"
elif not re.fullmatch(r"[0-9]+", amount_text):
reason = "invalid amount"
if reason:
errors.append({"record": number, "reason": reason})
continue
seen.add(key)
totals[category] += int(amount_text)
accepted += 1
with report.open("x", encoding="utf-8", newline="") as f:
writer = csv.writer(f)
writer.writerow(["category", "total"])
writer.writerows(sorted(totals.items()))
with rejected.open("x", encoding="utf-8", newline="") as f:
writer = csv.DictWriter(f, fieldnames=["record", "reason"])
writer.writeheader()
writer.writerows(errors)
print("accepted:", accepted, "rejected:", len(errors))
print("total:", sum(totals.values()))
raise SystemExit(1 if errors else 0)
同じフォルダーで実行します。
python csv_report.py
accepted: 3 rejected: 2、total: 14000が予想結果です。拒否があるので終了コードは1です。report.csvはfood 11500とtransport 2500、rejected.csvはduplicate idとinvalid amountを示すはずです。
再実行と不完全な出力を確認する
有効な入力の合計とレポートを照合します。別の新しいフォルダーで不正な2件を修正し、終了コード0を確認します。見出しの順序を変えた場合には、厳密な見出し規則の拒否も確認してください。
二つの出力作成は一つの原子的なトランザクションではなく、後半の失敗で一方だけ残る場合があります。実務では一時フォルダー、完了マーカー、配布前の検証などが必要です。不完全な出力を自動で後続処理へ渡さないでください。表計算ソフトへ渡す外部文字列は数式解釈にも対処します。引用や記録番号の前提はCSV形式の基本で確認できます。
次は提示した入力から得られる二つの結果ファイル全体です。受理した行の合計が正しくても、拒否行があれば入力全体の成功ではありません。終了値は1です。recordはヘッダーを1と数える論理レコード番号なので、重複は5、負数は6です。修正後は新しいフォルダーで実行し、上書き防止を回避しないでください。
report.csv
category,total food,11500 transport,2500
rejected.csv
record,reason 5,duplicate id 6,invalid amount
