毎月届くCSVを一つの表へまとめたいとき、先に確認したいのはファイル数より列の意味です。同じ売上という名前でも、税込と税抜、円と千円が混ざっていれば、そのまま足せません。まず同じ形式のファイルだけを結合します。
入力ファイルのそろえ方
この例は、列名と順番が同じ二つのCSVを使います。part-a.csvとpart-b.csvとして、同じ作業フォルダーへUTF-8で保存します。データは架空の例です。
店舗,金額
東京,1200店舗,金額
大阪,800対象をpart-*.csvに絞り、出力はmerged.csvにします。出力が入力の検索条件へ入らない名前にすると、次回、自分が作った結果をもう一度結合する誤りを避けられます。
列を確かめてから結合するコード
次をmerge_csv.pyへ保存して実行します。小さなCSVを対象に、全入力の検査が終わってから出力を作ります。
import csv
from pathlib import Path
expected = ["店舗", "金額"]
files = sorted(Path(".").glob("part-*.csv"))
if not files:
raise ValueError("入力CSVがありません")
rows = []
for path in files:
with path.open(encoding="utf-8-sig", newline="") as src:
reader = csv.DictReader(src)
if reader.fieldnames != expected:
raise ValueError(f"列名を確認してください: {path.name}")
for row in reader:
if set(row) != set(expected) or any(v is None for v in row.values()):
raise ValueError(f"項目数を確認してください: {path.name}")
rows.append(row)
with open("merged.csv", "x", encoding="utf-8-sig", newline="") as out:
writer = csv.DictWriter(out, fieldnames=expected)
writer.writeheader()
writer.writerows(rows)
print(f"入力{len(files)}ファイル・データ{len(rows)}行")画面には入力2ファイル・データ2行と出ます。merged.csvには見出しが一度だけ入り、その下に東京と大阪の行が並びます。既に出力があればFileExistsErrorで止まるので、内容を見て別の保存先を指定してください。
この段階では直さないもの
列名の順番が違う入力は停止させます。名前の揺れを推測して合わせる処理は入れていません。空文字の金額、数値の単位、同じ取引の重複も、このコードだけでは判断しません。結合と集計を分け、結合後の行数が入力の合計と一致するかを先に確かめます。
CSVの値は基本的に文字列として読まれます。金額を計算するときは、空欄やカンマの扱いを決めてから数値へ変換します。ファイル名で並べた順と、明細の日付順も同じとは限りません。並べ替えが必要なら、日付として読めることを別に確認します。
件数を増やす前の見直し
この例は全行をメモリーへ置きます。大きなデータでは入力を事前検査してから一時出力へ順次書く設計などを検討します。途中で止まった出力を完成品として配布しないよう、作成中と確認済みの場所を分けます。
月次運用では、使ったファイル名、入力行数、出力行数、対象月を記録します。前月分が混ざっていないか、同じファイルを重ねて受け取っていないかは、人が確認する項目として残します。
関連する作業
結合した後の集計と、欠損のチェックは次の記事で確認できます。
