PythonでCSVを自動集計する仕組み

毎朝CSVを開き、売上やポイントをコピーし、商品別の合計をExcelへ転記していませんか。

1回の作業は短くても、毎日繰り返せば自由な時間を削ります。しかも、担当者が休むと集計も止まります。この状態では、収益を確認するために人間が作業し続けなければなりません。

PythonでCSVを自動集計できるようになると、次の処理を人手から切り離せます。

  • 売上・ポイント・広告成果CSVの読み込み
  • 商品別、日付別、流入経路別の集計
  • 入力件数と合計金額の照合
  • 結果CSVと実行ログの保存
  • タスクスケジューラによる定期実行
  • 成果の低い商品や導線の抽出

この記事では、Pythonの標準機能だけを使い、CSVを安全に自動集計する基本パターンを解説します。読了後は、サンプルコードを手元で動かし、単発の時短ではなく、人間が介在しなくてもデータを整理し続ける自動化資産の土台を作れます。

ただし、集計プログラムを動かしただけで収益が発生するわけではありません。収益化には、需要のある商品、集客経路、適切な販売導線、規約を守った運用が必要です。本記事は一般的な情報提供であり、特定の収益やポイント獲得を保証するものではありません。

PythonによるCSV自動集計の全体像

CSVとは、表形式のデータを文字列として保存するファイル形式です。たとえば、次のような売上データがあります。

日付,商品名,流入元,売上金額
2026-07-21,商品A,検索,1200
2026-07-21,商品B,SNS,800
2026-07-22,商品A,検索,500

これを商品別に集計すると、次の結果になります。

商品名,売上合計
商品A,1700
商品B,800

PythonによるCSV自動集計は、次の5工程で構成します。

CSVを取得
  ↓
列名・金額・空欄を検証
  ↓
商品や流入元ごとに集計
  ↓
入力合計と出力合計を照合
  ↓
結果と実行ログを保存

収益改善につなげる場合は、この後ろに「改善判断」を接続します。

売上・ポイント・広告CSV
  ↓
Pythonで自動集計
  ↓
成果の高い商品・流入元を抽出
  ↓
記事、広告、商品ページの改善候補を作成
  ↓
クリック率・成約率・承認率を再計測

自動集計の価値は、合計を出すことに限りません。人が表計算ソフトを開かなくても、改善に使える情報が決まった時刻に届く状態を作れる点にあります。

10万行・20回で行ったCSV自動集計の検証ログ

一般論と実測結果を分けるため、Hiroの運営環境で匿名の合成CSVを生成し、本記事向けに集計処理を実行しました。

検証条件

実行日時：2026年7月23日 11:58:59 JST
OS：Windows 10.0.19045
Python：3.11.9
プロセッサー表記：AMD64 Family 23 Model 113
入力：合成CSV 100,000行
商品数：100種類
処理：商品別集計、CSV出力、出力再読込、合計照合
試行回数：20回
外部ライブラリ：なし

入力金額は、行番号に基づく規則で生成したテスト値です。実際の売上やポイント実績ではありません。

検証結果

入力行数：100,000
出力行数：100
入力合計：49,795,450
出力合計：49,795,450
処理時間中央値：180.959ミリ秒
最小：173.286ミリ秒
最大：244.770ミリ秒
入力ファイル容量：989,118バイト
SHA-256：
75a4325e5fd4c44a85ad236c52e9b54100d16c9af8facd0334e9721213c57dcd

20回すべてで、入力合計と出力合計の一致を確認しています。

この結果は、上記環境と合成データを使ったローカル測定です。別のPC、ネットワークドライブ、ウイルス対策ソフト、ストレージ性能、CSVの文字コードによって処理時間は変わります。また、人間の手作業時間とは測定条件が異なるため、この結果だけから「人より何倍速い」とは判断できません。

ここから確認できるのは、10万行を100商品へ集約し、出力を再読込したうえで金額を照合できたという処理結果です。なお、後述する学習用サンプルコードは安全な集計構造を理解するためのもので、上記ベンチマークとまったく同じ測定条件を再現するコードではありません。

CSV集計前後と検証ログの比較図

ステップ・バイ・ステップ：PythonでCSVを自動集計する手順

1. 自動化するCSVを1つ選ぶ

最初の対象には、手順と正解を数字で説明できるCSVを選びます。

適している例は次のとおりです。

  • アフィリエイト成果の提供元別集計
  • ECサイトの売上商品別集計
  • ポイント履歴のサービス別集計
  • 広告費とコンバージョンの媒体別集計
  • ブログ記事のクリック数・成約数集計

対象を決めたら、現在の作業を記録します。

作業名：
実行頻度：
1回の実測時間：
入力ファイル：
集計単位：
出力ファイル：
正しい結果を判定する数字：

削減効果は、実測した時間から計算します。

月間削減時間
＝（従来の作業時間－自動化後の確認時間）×月間実行回数

たとえば、手作業が実測20分、自動化後の確認が4分、月20回なら、前提上の削減時間は320分です。ただし、導入時の開発時間と継続的な保守時間も差し引いて評価してください。

2. 入力CSVの仕様を確認する

コードを書く前に、CSVの列名と業務ルールを決めます。

今回の例では、次の形式を使います。

商品名,売上金額
商品A,1200
商品B,800
商品A,500

確認項目は次のとおりです。

  • 文字コードはUTF-8、UTF-8 with BOM、CP932のどれか
  • 区切り文字はカンマかタブか
  • 金額は税込みか税抜きか
  • 返金を負数で表すか
  • 空欄をエラーにするか
  • 商品名の前後に空白が入る可能性があるか
  • 注文IDなど、重複判定に使える列があるか
  • 同名商品を同一商品として扱ってよいか
  • 小数や端数をどの単位で丸めるか

仕様が曖昧なまま自動化すると、プログラムが正常終了しても、事業上は誤った数字になる恐れがあります。

3. 作業フォルダを分ける

次の構成にすると、元データと生成物を混同しにくくなります。

csv-automation/
├─ aggregate_sales.py
├─ input/
│  └─ sales.csv
├─ output/
└─ logs/

入力CSVへ直接書き込む設計は避けます。途中で処理が止まると、元データまで壊す可能性があるためです。

4. 集計スクリプトを作る

aggregate_sales.pyとして次のコードを保存します。

import csv
import json
import os
from collections import defaultdict
from datetime import datetime
from decimal import Decimal, InvalidOperation
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent
INPUT_PATH = BASE_DIR / "input" / "sales.csv"
OUTPUT_PATH = BASE_DIR / "output" / "summary.csv"
TEMP_PATH = BASE_DIR / "output" / "summary.csv.tmp"
LOG_PATH = BASE_DIR / "logs" / "run.jsonl"

REQUIRED_COLUMNS = {"商品名", "売上金額"}

for directory in (OUTPUT_PATH.parent, LOG_PATH.parent):
    directory.mkdir(parents=True, exist_ok=True)


def write_log(status, **values):
    record = {
        "run_at": datetime.now().astimezone().isoformat(),
        "status": status,
        **values,
    }
    with LOG_PATH.open("a", encoding="utf-8") as file:
        file.write(json.dumps(record, ensure_ascii=False) + "\n")


try:
    if not INPUT_PATH.exists():
        raise FileNotFoundError(f"入力CSVがありません: {INPUT_PATH}")

    totals = defaultdict(Decimal)
    input_rows = 0
    input_total = Decimal("0")

    with INPUT_PATH.open(
        encoding="utf-8-sig",
        newline="",
    ) as file:
        reader = csv.DictReader(file)

        if reader.fieldnames is None:
            raise ValueError("ヘッダーがありません")

        missing = REQUIRED_COLUMNS - set(reader.fieldnames)
        if missing:
            raise ValueError(f"必要な列がありません: {sorted(missing)}")

        for line_number, row in enumerate(reader, start=2):
            product_raw = row.get("商品名")
            amount_raw = row.get("売上金額")

            if product_raw is None:
                raise ValueError(f"{line_number}行目の商品名を読み取れません")
            if amount_raw is None:
                raise ValueError(f"{line_number}行目の金額を読み取れません")

            product = product_raw.strip()
            amount_text = amount_raw.replace(",", "").strip()

            if not product:
                raise ValueError(f"{line_number}行目の商品名が空です")
            if not amount_text:
                raise ValueError(f"{line_number}行目の金額が空です")

            try:
                amount = Decimal(amount_text)
            except InvalidOperation as error:
                raise ValueError(
                    f"{line_number}行目の金額が不正です: {amount_text}"
                ) from error

            if not amount.is_finite():
                raise ValueError(
                    f"{line_number}行目の金額が有限値ではありません"
                )

            totals[product] += amount
            input_total += amount
            input_rows += 1

    if input_rows == 0:
        raise ValueError("入力データが0件です")

    with TEMP_PATH.open(
        "w",
        encoding="utf-8-sig",
        newline="",
    ) as file:
        writer = csv.writer(file)
        writer.writerow(["商品名", "売上合計"])

        for product, total in sorted(totals.items()):
            writer.writerow([product, str(total)])

    output_total = Decimal("0")
    output_rows = 0

    with TEMP_PATH.open(encoding="utf-8-sig", newline="") as file:
        for row in csv.DictReader(file):
            output_total += Decimal(row["売上合計"])
            output_rows += 1

    if input_total != output_total:
        raise RuntimeError("入力合計と出力合計が一致しません")

    os.replace(TEMP_PATH, OUTPUT_PATH)

    write_log(
        "ok",
        input_rows=input_rows,
        output_rows=output_rows,
        input_total=str(input_total),
        output_total=str(output_total),
    )

    print(f"status=ok input_rows={input_rows}")
    print(f"input_total={input_total} output_total={output_total}")

except Exception as error:
    if TEMP_PATH.exists():
        TEMP_PATH.unlink()

    write_log(
        "error",
        error_type=type(error).__name__,
        error_message=str(error),
    )
    raise

csv.DictReaderは、列を位置ではなく名前で読む機能です。たとえば、row["売上金額"]のように指定できるため、列の並び替えに比較的強くなります。

金額にはfloatではなくDecimalを使っています。Decimalは10進数を扱う型で、金額計算における二進浮動小数点の誤差を避けやすくします。

また、結果をいきなりsummary.csvへ書かず、一時ファイルへ保存して検証した後、os.replace()で置き換えています。処理途中で失敗しても、不完全な出力を正式な集計結果として残しにくくするためです。

このサンプルが想定する入力文字コードはUTF-8またはUTF-8 with BOMです。CP932のCSVを使う場合は、提供元の仕様を確認したうえでencoding="cp932"へ変更してください。

5. 手動実行して結果を検証する

PowerShellで作業フォルダへ移動し、次を実行します。

py aggregate_sales.py

サンプルCSVなら、次の結果が期待値です。

status=ok input_rows=3
input_total=2500 output_total=2500

output/summary.csvも確認します。

商品名,売上合計
商品A,1700
商品B,800

成功判定には、次の条件を使います。

  • 必須列が存在する
  • 入力行数が0件ではない
  • 商品名や金額に不正値がない
  • 出力ファイルを再読込できる
  • 入力合計と出力合計が一致する
  • ログに"status": "ok"が記録される

「エラーが表示されなかった」だけでは、集計結果の正しさを証明できません。書き出したファイルを再度読み、合計を照合するところまで自動化します。

6. 異常データをテストする

次のCSVも試してください。

商品名,売上金額
商品A,1200
商品B,金額不明

処理は成功せず、ログへ次のような情報が残る設計です。

{
  "status": "error",
  "error_type": "ValueError",
  "error_message": "3行目の金額が不正です: 金額不明"
}

実際のrun.jsonlでは、上記の情報に実行日時が加わり、1実行につき1行のJSONとして保存されます。

ほかにも、次のケースを確認します。

  • CSVが存在しない
  • ヘッダーしかない
  • 売上金額列がない
  • 商品名が空欄
  • 金額がNaNやInfinity
  • 同じ注文IDが複数回登場する
  • Excelで出力CSVを開いたままにする
  • CSVの途中に列数が足りない行がある
  • 返金を表す負数が含まれている

異常時に誤った集計結果を残さないことが、無人運用の前提です。

なお、このサンプルコードには注文IDによる重複排除は含まれていません。同じCSVを繰り返し取り込む運用へ進む前に、取引IDやファイル識別子を使った重複防止を追加してください。

7. 定期実行と通知を設定する

Windowsなら、タスクスケジューラで実行できます。

プログラム：
C:\automation\csv-automation\.venv\Scripts\python.exe

引数：
C:\automation\csv-automation\aggregate_sales.py

開始：
C:\automation\csv-automation

複数のPython環境がある場合は、pythonという短い指定ではなく、仮想環境内の実行ファイルを絶対パスで設定します。

定期実行へ移す前に、少なくとも次の点を確認してください。

  • 手動実行で数回連続して成功する
  • 異常なCSVでは処理が失敗する
  • タスクの実行ユーザーが入力・出力フォルダへアクセスできる
  • 終了コードが0以外なら検知できる
  • 前回の成功時刻を確認できる
  • ログが無制限に増え続けない

完全自動化へ進めるなら、次の処理も追加候補です。

  • CSVの自動取得
  • エラー時のメール・チャット通知
  • 前日比が一定範囲を超えた場合の警告
  • 成果上位商品の自動抽出
  • 集計結果のダッシュボード反映
  • 商品ページや記事改善の候補リスト作成

自動申込、自動クリック、ポイント獲得操作などは、サービス規約違反や成果否認につながる可能性があります。正当に取得したCSVの整理・分析・通知を中心に設計してください。

専門家目線のチェックポイント

金額の意味を統一する

「売上」には、発生額、承認待ち額、承認額、返金控除後の金額などが混在します。

アフィリエイトやポイントを扱うなら、少なくとも次の状態を分けます。

発生額
承認待ち額
承認額
否認額
返金・取消額

見かけの発生額をそのまま収益として扱うと、実際の入金額とずれる可能性があります。列名にも売上のような曖昧な名称ではなく、承認済み報酬額など、業務上の意味が分かる名称を使う方が安全です。

重複取り込みを防ぐ

同じCSVを2回処理すると、売上が二重計上されることがあります。注文IDや成果IDがあるなら、提供元＋取引IDのような一意キーを記録します。

日付と金額だけの重複判定では、同日に発生した正当な同額取引まで削除する恐れがあります。

実務では、次のいずれかを保存します。

  • 処理済み取引ID
  • 処理済みファイルのSHA-256
  • 提供元のレポートID
  • 取り込み対象期間と実行日時
  • 元CSVのファイル名と更新日時

ファイルのハッシュが同じでも、同一内容の再処理を許可したい業務はあります。重複の定義は、コードを書く前に業務ルールとして決めてください。

エラー行を無視するか停止するか決める

広告レポートなら、不正行だけを隔離して処理を続ける判断もあり得ます。一方、請求や会計に関係するCSVでは、1行の欠損でも全体を停止した方が安全です。

判断基準は次の3点です。

  • 欠損が金額へ与える影響
  • 後から修正できるか
  • 誤集計した場合の損害

処理を続ける場合は、不正行の件数だけでなく、行番号、理由、元データの識別子を隔離ファイルへ保存します。個人情報や機密情報をログへそのまま記録しない配慮も必要です。

集計軸の表記揺れを管理する

商品A、商品Ａ、商品A は、人間には同じ商品に見えても、Python上では別の文字列になる場合があります。

今回のコードは前後の空白だけを除去します。全角・半角、大文字・小文字、旧商品名などの表記揺れは統合しません。

本番運用では、次のような変換表を別途管理します。

入力商品名,正規商品名
商品Ａ,商品A
旧プラン名,新プラン名

自動的な文字変換を増やしすぎると、別商品を誤って統合する恐れがあります。正規化ルールは明示し、変更履歴を残してください。

ネットワークドライブを過信しない

タスクスケジューラの実行ユーザーから、ネットワークドライブが見えない場合があります。また、クラウド同期中に一時ファイルを置換すると、ローカルディスクとは異なる挙動になる可能性があります。

入力パス、出力パス、実行ユーザー、同期状態をログへ残すと、原因を追跡しやすくなります。

重要な処理では、いったんローカルディスクへコピーして集計し、検証後の結果だけを同期先へ配置する設計も検討してください。

同時実行を防ぐ

前回の処理が終わる前に次の処理が始まると、同じ一時ファイルや出力ファイルを同時に更新する可能性があります。

今回のコードには、ロックファイルや排他制御がありません。定期実行の間隔が短い場合や、処理時間が変動する場合は、次の対策が必要です。

  • タスクスケジューラで多重起動を禁止する
  • 実行中を示すロックファイルを作る
  • 実行ごとに異なる一時ファイル名を使う
  • SQLiteなどで処理状態を管理する

視覚的証拠として追加したい画像

記事へ視覚的証拠を追加するなら、次の画像が理解を助けます。

  1. 入力CSVと出力CSVの比較図
    左側に3行の売上明細、右側に商品別集計を置き、商品Aの2行が1行へ合算される様子を同じ色で示します。

  2. 処理フロー図
    取得→検証→集計→照合→保存→通知を矢印で表現します。異常時は正式な出力を更新せず、エラーログへ分岐する線も加えます。

  3. 実行ログのスクリーンショット
    実行日時、入力行数、出力行数、入力合計、出力合計、ステータスを表示します。ユーザー名、注文ID、APIキーなどは必ず伏せます。

生成された概念画像は仕組みの説明には使えますが、実行の証拠にはなりません。実運用の記事では、匿名化したターミナル画面、実行ログ、入力・出力ファイルのハッシュなども併記すると、読者が再現性を判断しやすくなります。

よくある失敗と対策

FileNotFoundErrorが出る

実行中のフォルダとCSVの保存場所がずれている可能性があります。

print(INPUT_PATH.resolve())
print(INPUT_PATH.exists())

実際に参照している絶対パスを表示して確認します。

今回のコードは__file__を基準にパスを作るため、PowerShellでどのフォルダから実行しても、原則としてスクリプトと同じ場所にあるinput、output、logsを参照します。

日本語が文字化けする

Excel由来のCSVなら、まずutf-8-sigを試します。提供元がShift_JIS互換形式を指定している場合はcp932を使います。

INPUT_PATH.open(encoding="cp932", newline="")

文字コードを推測で切り替えるのではなく、提供元の仕様を確認してください。同じ処理の途中で複数の文字コードを無条件に試すと、誤った文字として読めてもエラーにならない場合があります。

金額の変換に失敗する

9,800円や¥9,800など、数字以外の文字が入っている可能性があります。許可する書式を先に決め、対象外の形式はエラーとして記録します。

サンプルコードはカンマだけを除去します。円記号や単位を自動削除しないのは、想定外の形式を黙って受け入れないためです。

なお、厳格な金額検証が必要なら、1,2,3のような不正なカンマ位置も受け入れないよう、変換前に正規表現で書式を確認してください。

列名が突然変わる

提供元の仕様変更が考えられます。列番号で読み続けず、必須列を起動時に検査し、欠けていれば停止して通知します。

列名の前後に空白が加わっただけでも別の列名として扱われます。勝手に空白を削除して処理を続けるか、仕様変更として停止するかは、データ提供元の運用に合わせて決めます。

Excelで出力CSVを開いていると失敗する

Windowsでは、Excelが出力ファイルをロックしていると、os.replace()がPermissionErrorで失敗することがあります。

この場合、一時ファイルは削除され、既存のsummary.csvは更新されません。Excelを閉じて再実行してください。無人運用では、出力先を実行日時付きのファイル名にする方法もあります。

自動実行では動かない

手動実行と定期実行では、ユーザー、作業フォルダ、環境変数、アクセス権限が異なる場合があります。Python、スクリプト、入力CSVを絶対パスで指定し、ログにも実際のパスを残します。

タスクスケジューラでは「ユーザーがログオンしているかどうかにかかわらず実行する」という設定によって、見えるドライブや認証状態が変わることがあります。手動実行の成功だけで定期実行も成功すると判断しないでください。

成果を測るKPI

KPI計算・記録方法改善に使う判断
実行成功率成功回数÷予定実行回数定期実行の安定性を見る
正常処理率正常行数÷入力行数CSV仕様変更やデータ品質を見る
合計一致率合計一致回数÷実行回数集計の整合性を見る
確認時間人間が結果確認に使った実測時間監視方法を見直す
純削減時間従来時間－確認・保守時間自動化の費用対効果を見る
承認率承認件数÷発生件数案件や流入元の質を見る
商品別CVR購入数÷商品ページ訪問数記事と商品の適合を改善する
CTAクリック率/products/クリック数÷記事閲覧数収益導線の接続を評価する

目標値は他サイトの数字を借りず、導入前後の実測から決めます。最初の期間は、人間による確認時間、エラー件数、保守時間も記録してください。

たとえば、月間削減時間だけが増えても、誤集計の調査に長時間かかっていれば成功とはいえません。時間、正確性、安定性を分けて測ることが重要です。

この方法の限界と使えないケース

PythonによるCSV自動集計が適さない場面もあります。

  • 1回しか行わない小規模な集計
  • 毎回、列や計算ルールが変わるデータ
  • 数字だけでは正解を判定できない業務
  • 本人確認や手動承認が必要なデータ取得
  • 自動アクセスが規約で禁止されているサービス
  • 複数人がリアルタイムで同時更新するデータ
  • 厳格な監査証跡や権限分離が必要な会計処理
  • 数百万行以上を継続的に分析する用途

数十行を一度だけ合計するなら、Excelのピボットテーブルの方が早い場合があります。データ量や結合条件が増えたら、pandas、DuckDB、SQLite、データベースの利用も検討します。

今回のサンプルには、次の機能が含まれていません。

  • 外貨換算
  • 税計算と端数処理
  • 返金と取消の業務ルール
  • 処理済みIDの永続管理
  • 自動通知
  • ログの世代管理と自動削除
  • 同時実行制御
  • 個人情報のマスキング
  • 電子帳簿保存法などへの対応
  • 会計システムとの照合

本番では、対象業務の重要度と誤集計時の損害に応じた追加設計が必要です。特に会計・税務判断へ利用する場合は、担当者や専門家による確認を省略しないでください。

類似記事との差別化ポイント

一般的なPython・CSV記事は、ファイルを読み、合計を画面表示するところで終わりがちです。

本記事では、次の運用範囲まで扱いました。

  • 10万行の合成CSVによる20回の実測
  • 測定条件、実測値、限界の明記
  • 必須列、空欄、不正値の検査
  • Decimalによる金額処理
  • 出力CSVの再読込と合計照合
  • 一時ファイルを使った安全な置換
  • 成功・失敗ログの保存
  • 定期実行時の環境差への対策
  • 発生額と承認額を分ける収益管理
  • 重複取り込み、表記揺れ、同時実行のリスク整理
  • 集計結果を商品・記事・CTA改善へ戻す設計

単なるPythonの練習用コードではなく、継続運用できる収益管理基盤へ育てる入口として設計している点が違いです。

まとめ：今日から始める具体的アクション

読了後は、過去7日分の売上、ポイント、広告成果、作業記録のうち、どれか1種類をCSVへまとめてください。

次の順番で進めます。

  1. 集計したいCSVを1つ選ぶ
  2. 列名と集計ルールを文章にする
  3. 入力CSVのコピーを使ってサンプルコードを手動実行する
  4. 入力合計と出力合計を照合する
  5. 異常データで安全に停止するか試す
  6. 実行時間と確認時間を記録する
  7. 数回の照合後に定期実行へ移す
  8. 集計結果を商品や集客導線の改善に使う

最初のアクションは、次の4項目をメモすることです。

現在の手作業時間：
月間実行回数：
集計対象のCSV：
正しい結果を証明できる数字：

いきなり本番データを無人処理する必要はありません。まずは元CSVのコピーを使い、手作業の結果とPythonの結果が一致するかを確認してください。

これが、自分の時間を切り売りする作業から、繰り返し働く自動化資産へ移行する出発点になります。

本気で自動化・収益基盤を構築したい方向けの実践マニュアル

CSV集計を自動化できれば、数字を集める作業から離れられます。しかし、収益につながる仕組みを作るには、データ取得、条件判定、コンテンツ生成、商品販売、KPI改善までを一つの流れとして接続する必要があります。

Hiroの実践マニュアルでは、AIブログ、アフィリエイト、デジタル商品、VPS運用などを題材に、自分が作業していない時間にも価値提供と収益導線が動き続ける仕組みを、実装単位で解説しています。

毎回CSVを開いて結果を追いかける側から、データと改善候補が自動で集まり、収益導線を育てられる側へ移りたい方は、次のページから目的に合う実践ルートを選んでください。

👉 本気で自動化・収益基盤を構築するための実践マニュアルを見る