PythonでCSVを自動集計する基本パターン：手作業レポートを「改善に使える自動化資産」に変える入門

date,channel,revenue,points
2026-07-01,blog,120,1
2026-07-01,room,0,3
2026-07-02,mail,320,0

python=3.11.9
platform=Windows-10-10.0.19045-SP0
rows=1200
elapsed_ms=2.564
blog {'rows': 300, 'revenue': 34200, 'points': 300}
mail {'rows': 300, 'revenue': 34200, 'points': 300}
room {'rows': 300, 'revenue': 34200, 'points': 1800}
sns {'rows': 300, 'revenue': 34200, 'points': 1800}

date,channel,revenue,points,status
2026-07-01,blog,120,1,approved
2026-07-01,room,0,3,pending
2026-07-02,mail,320,0,approved
2026-07-02,sns,80,10,rejected

channel,rows,approved_revenue,pending_revenue,rejected_rows,points
blog,1,120,0,0,1
room,1,0,0,0,3
mail,1,320,0,0,0
sns,1,0,0,1,10

date
channel
revenue
points
status

import csv

with open("sales.csv", encoding="utf-8", newline="") as f:
    reader = csv.DictReader(f)

    for row in reader:
        print(row)

2026-07-01,blog,120,1,approved

row["channel"]
row["revenue"]
row["status"]

import csv

required_columns = {"date", "channel", "revenue", "points", "status"}

with open("sales.csv", encoding="utf-8", newline="") as f:
    reader = csv.DictReader(f)
    actual_columns = set(reader.fieldnames or [])

    missing = required_columns - actual_columns
    if missing:
        raise ValueError(f"必要な列がありません: {sorted(missing)}")

revenue = row["revenue"]

revenue = int(row["revenue"])
points = int(row["points"])

def to_int(value):
    if value is None:
        return 0

    value = str(value).strip()

    if value in {"", "-", "N/A", "null"}:
        return 0

    return int(value)

import csv
from collections import defaultdict

required_columns = {"date", "channel", "revenue", "points", "status"}

summary = defaultdict(lambda: {
    "rows": 0,
    "approved_revenue": 0,
    "pending_revenue": 0,
    "rejected_rows": 0,
    "points": 0,
})

def to_int(value):
    if value is None:
        return 0

    value = str(value).strip()

    if value in {"", "-", "N/A", "null"}:
        return 0

    return int(value)

with open("sales.csv", encoding="utf-8", newline="") as f:
    reader = csv.DictReader(f)

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

    for row in reader:
        channel = row["channel"].strip()
        status = row["status"].strip().lower()
        revenue = to_int(row["revenue"])
        points = to_int(row["points"])

        summary[channel]["rows"] += 1
        summary[channel]["points"] += points

        if status == "approved":
            summary[channel]["approved_revenue"] += revenue
        elif status == "pending":
            summary[channel]["pending_revenue"] += revenue
        elif status == "rejected":
            summary[channel]["rejected_rows"] += 1
        else:
            raise ValueError(f"未知のstatusです: {status}")

for channel, values in summary.items():
    print(channel, values)

with open("summary.csv", "w", encoding="utf-8", newline="") as f:
    fieldnames = [
        "channel",
        "rows",
        "approved_revenue",
        "pending_revenue",
        "rejected_rows",
        "points",
    ]

    writer = csv.DictWriter(f, fieldnames=fieldnames)
    writer.writeheader()

    for channel, values in sorted(summary.items()):
        writer.writerow({
            "channel": channel,
            "rows": values["rows"],
            "approved_revenue": values["approved_revenue"],
            "pending_revenue": values["pending_revenue"],
            "rejected_rows": values["rejected_rows"],
            "points": values["points"],
        })

import csv
import time
from collections import defaultdict
from pathlib import Path

INPUT_PATH = Path("sales.csv")
OUTPUT_PATH = Path("summary.csv")
ENCODING = "utf-8"

REQUIRED_COLUMNS = {"date", "channel", "revenue", "points", "status"}

def to_int(value):
    if value is None:
        return 0

    value = str(value).strip()

    if value in {"", "-", "N/A", "null"}:
        return 0

    return int(value)

def main():
    start = time.perf_counter()

    summary = defaultdict(lambda: {
        "rows": 0,
        "approved_revenue": 0,
        "pending_revenue": 0,
        "rejected_rows": 0,
        "points": 0,
    })

    with INPUT_PATH.open(encoding=ENCODING, newline="") as f:
        reader = csv.DictReader(f)

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

        input_rows = 0

        for row in reader:
            input_rows += 1

            channel = row["channel"].strip()
            status = row["status"].strip().lower()
            revenue = to_int(row["revenue"])
            points = to_int(row["points"])

            if not channel:
                raise ValueError(f"channelが空です: row={input_rows}")

            summary[channel]["rows"] += 1
            summary[channel]["points"] += points

            if status == "approved":
                summary[channel]["approved_revenue"] += revenue
            elif status == "pending":
                summary[channel]["pending_revenue"] += revenue
            elif status == "rejected":
                summary[channel]["rejected_rows"] += 1
            else:
                raise ValueError(f"未知のstatusです: row={input_rows}, status={status}")

    with OUTPUT_PATH.open("w", encoding="utf-8", newline="") as f:
        fieldnames = [
            "channel",
            "rows",
            "approved_revenue",
            "pending_revenue",
            "rejected_rows",
            "points",
        ]

        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()

        for channel, values in sorted(summary.items()):
            writer.writerow({
                "channel": channel,
                "rows": values["rows"],
                "approved_revenue": values["approved_revenue"],
                "pending_revenue": values["pending_revenue"],
                "rejected_rows": values["rejected_rows"],
                "points": values["points"],
            })

    elapsed_ms = (time.perf_counter() - start) * 1000

    print(f"input={INPUT_PATH}")
    print(f"output={OUTPUT_PATH}")
    print(f"rows={input_rows}")
    print(f"channels={len(summary)}")
    print(f"elapsed_ms={elapsed_ms:.3f}")

if __name__ == "__main__":
    main()

UnicodeDecodeError: 'utf-8' codec can't decode byte...

with open("sales.csv", encoding="cp932", newline="") as f:
    ...

data/raw/sales_2026-07-12.csv
data/output/summary_2026-07-12.csv
logs/run_2026-07-12.txt

approved_revenue = 確定収益
pending_revenue = 見込み収益
rejected_rows = 否認件数

required = {"date", "channel", "revenue", "points", "status"}
actual = set(reader.fieldnames or [])

missing = required - actual
if missing:
    raise ValueError(f"必要な列がありません: {sorted(missing)}")

def to_int(value):
    if value is None:
        return 0

    value = str(value).strip()

    if value in {"", "-", "N/A", "null"}:
        return 0

    return int(value)

processed_files.txt
sales_2026-07-11.csv
sales_2026-07-12.csv

input=sales.csv
output=summary.csv
rows=1200
channels=4
elapsed_ms=2.564
status=success

status=failed
error=必要な列がありません: ['revenue']
input=sales.csv

1回15分 × 週3回 = 週45分
週45分 × 4週 = 月180分

対象CSV: room_points_2026-07.csv
行数: 850行
毎回見る数字: 商品別ポイント、投稿日別クリック、承認済みポイント
手作業時間: 1回あたり約15分
自動化したい出力: summary_room_points.csv
必須列: date, product_id, clicks, points, status
注意点: pendingとapprovedを分ける

1ファイル
1集計
1出力
1ログ