【Python】ClickでCSVを集計するCLIツールを作ってみた

CSVファイルの集計を、表計算ソフトを開かずにコマンド一発で終わらせたい——そんな場面は意外と多いのではないでしょうか。今回はPythonのClickと標準ライブラリのcsvだけで、売上CSVを集計する小さなコマンドラインツールcsvaggを作ってみました。

sample・summary・groupby・filterの4コマンドを備え、数値列の自動判定から集計、絞り込み、CSV出力までをこなせます。初心者でも追体験できるように、実装の要点と実行の様子を順番に解説していきます。

Clickの基本概念、要件定義、実装、動作確認までを順番に学べる構成です。動画は目次から確認したい場面へ移動でき、本文だけでも手順と考え方が完結します。

今回使用する主要なライブラリについて、役割と使い分けを順番に確認します。

CLIを組み立てるライブラリ「Click」の仕組み

Clickは、普通のPython関数をコマンドライン用のサブコマンドへ変えてくれるライブラリです。関数の上に@click.commandや@click.optionといったデコレータを重ねるだけで、引数の解釈やヘルプ表示、型チェックまで任せられます。

今回のcsvaggでは@click.groupで4つのコマンドを束ね、pathやoptionを受け取る入口として活用しました。細かな引数処理を自前で書かずに済むので、集計そのものへ集中できます。

• 関数をデコレータでサブコマンドに変える
• @click.groupで複数コマンドを束ねる
• @click.argumentで必須の引数を受け取る
• @click.optionで任意のオプションを足す
• ヘルプや使い方を自動で生成する

CSVを読み書きする標準ライブラリ「csv」の役割

csvは、CSVファイルの読み書きを担うPython標準ライブラリで、追加インストールなしに使えます。なかでもDictReaderは各行をヘッダー名をキーにした辞書として返すため、列名で値を取り出せて便利です。書き出しにはDictWriterを用い、ヘッダー行と本体をまとめて出力します。

文字コードにutf-8-sigを選んでいる点も大切で、日本語を含むCSVをExcelで開いても文字化けを避けられます。

• DictReaderで行を辞書として読む
• DictWriterで辞書をCSVに書く
• fieldnamesでヘッダーを取得する
• utf-8-sigで文字化けを防ぐ

この記事のセットアップ手順と掲載コードは、Windows 11 Pro、PowerShell 5.1、Python 3.13.3で動作確認しています。仮想環境を有効化せず、その中のPythonを直接指定するため、以下のコマンドはPowerShellとコマンドプロンプト（cmd）の両方で使えます。

python -m venv .venv
.\.venv\Scripts\python.exe -m pip install click==8.1.7

macOS・Linuxでは仮想環境内のPythonパスが異なります。今回の動作確認環境とは異なるため、以下は環境差分を補う参考手順です。

python3 -m venv .venv
./.venv/bin/python -m pip install click==8.1.7

• CSVはutf-8-sig(BOM付きUTF-8)で読み書きするため、Excelで開いても日本語が文字化けしません。
• サブコマンド型のツールです。--helpを付けて実行すると、使えるコマンドの一覧を確認できます。

目的は、PythonのClickと標準csvライブラリで売上CSVを集計するCLIツールを作れるようになることです。

対象者として、PythonでコマンドラインツールとCSV処理の基礎を学びたい人を想定しています。

完成物は、sample・summary・groupby・filterの4コマンドを備えたCSV集計CLIツールcsvaggです。

実装へ入る前に、機能・品質・受け入れ条件を分けて確認します。

機能要件

• サンプル売上CSVを書き出す
• 数値列を自動判定して集計する
• 件数・合計・平均・最小・最大を求める
• 指定した列でグループ化して集計する
• 指定した列と値で行を絞り込む
• 集計や絞り込みの結果をCSVへ保存する

非機能要件

• 追加ライブラリはClickだけで完結する
• utf-8-sigで日本語の文字化けを防ぐ
• 不正な列名は分かりやすいエラーで伝える
• 固定データで実行例と手元の出力を一致させる

実装方針

今回はClickと標準csvの基本動作を追いやすくするため、CSV集計CLIツール本体を1つのPythonファイルへまとめます。

入力、判定、結果表示の役割を分け、実行結果を確認しながら機能を積み上げます。

• Clickと標準csvのコマンド定義と処理関数の責務を分ける
• 入力フォルダやオプションを検証してからファイル処理を始める
• 時間のかかる処理は進捗を表示し、結果を確認できるようにする
• 削除ではなく退避を選び、誤判定から戻せる設計にする

完成と判断する条件

1. sampleで12件の売上CSVが作られる
2. summaryで数量と金額が集計される
3. groupbyでカテゴリ別の合計が出る
4. filterで一致件数と該当行が表示される
5. 出力指定時に結果CSVが保存される

csvaggの心臓部は、数値列の自動判定と集計処理のまとまりです。numeric_columnsは各列の値をすべてfloatへ変換できるか試し、成功した列だけを集計対象として拾い上げます。実際の集計はaggregate関数へ集約し、合計や平均などをhowという引数で切り替えられるようにしました。

列の並び順はOrderedDictで保ち、整数はfmtで小数点を落として読みやすく表示します。

• 全行が数値の列だけを対象にする
• sumやmeanを1つの関数で切り替える
• OrderedDictで列の順序を保つ
• 整数は小数点を省いて表示する

集計する列を自動で見分ける方法

numeric_columnsは、列ごとに空でない値を集め、すべてfloatへ変換できるかを試します。1つでも変換に失敗した列は集計対象から外れるため、列名を決め打ちしなくても数値列だけを選び出せます。

• 列ごとに全値のfloat変換を試す
• 変換に失敗した列は対象から外す
• 列名を決め打ちせず柔軟に選ぶ

集計方法を1か所へまとめる工夫

合計や平均といった計算は、aggregate関数へ集約しています。howという引数で処理を切り替えられるので、summaryもgroupbyも同じ関数を呼ぶだけで済むのです。未知の方法が渡されたときには、ClickExceptionで知らせます。

• count・sum・mean・min・maxを共通化
• howで集計方法を切り替える
• 未知の方法は例外で知らせる

グループ化と絞り込みの考え方

groupbyでは、setdefaultを使ってグループごとに数値を集め、まとめて集計します。filterのほうは、指定した列の値が一致する行だけを残す仕組みです。どちらも画面への表示に加えて、出力指定でCSVへ保存できるようにしています。

• setdefaultでグループごとに値を集める
• 指定列と値が一致する行を残す
• 表示と保存の両方に対応する

ここからはcsvaggの中身を見ていきます。ファイルは大きく分けて、デモデータの定義、CSVの入出力、数値列の判定、集計処理、コマンド定義の5つで構成しています。

集計のロジックはaggregateという1つの関数にまとめました。summaryとgroupbyの両方から呼び出すので、同じ計算を二重に書かずに済みます。

以下では、初心者がまず押さえたい要点だけを抜き出して順番に解説します。

"""csvagg: ClickでCSVを集計する小さなCLIツール。

サブコマンド:
  sample   デモ用の売上CSVを書き出す
  summary  数値列を自動判定して統計量を集計する
  groupby  指定した列でグループ化して数値列を集計する
  filter   指定した列の値で行を絞り込む
"""

import csv
from collections import OrderedDict

import click


# デモデータの定義
# 固定の売上データ。毎回同じ内容を書き出すので、記事の実行例と手元の出力が一致する。
SAMPLE_HEADER = ["日付", "カテゴリ", "商品名", "数量", "金額"]
SAMPLE_ROWS = [
    ["2024-01-05", "家電", "電気ケトル", 2, 7600],
    ["2024-01-08", "食品", "オーガニック珈琲", 5, 4500],
    ["2024-01-11", "家電", "ヘッドホン", 1, 12800],
    ["2024-01-15", "文具", "万年筆", 3, 9600],
    ["2024-01-19", "食品", "紅茶ギフト", 4, 6400],
    ["2024-01-22", "家電", "加湿器", 2, 11800],
    ["2024-01-26", "文具", "ノート5冊組", 6, 3000],
    ["2024-02-02", "食品", "オーガニック珈琲", 3, 2700],
    ["2024-02-06", "家電", "電気ケトル", 1, 3800],
    ["2024-02-09", "文具", "ボールペン", 10, 2000],
    ["2024-02-14", "食品", "紅茶ギフト", 2, 3200],
    ["2024-02-18", "家電", "ヘッドホン", 2, 25600],
]


# CSVの入出力
# 読み書きともに utf-8-sig を使い、Excelで開いても日本語が文字化けしないようにする。
def load_csv(path):
    """CSVを読み込み、ヘッダーと行(dictのリスト)を返す。"""
    with open(path, newline="", encoding="utf-8-sig") as f:
        reader = csv.DictReader(f)
        rows = list(reader)
    return reader.fieldnames or [], rows


def save_csv(path, header, rows):
    """dictのリストを utf-8-sig のCSVとして書き出す。"""
    with open(path, "w", newline="", encoding="utf-8-sig") as f:
        writer = csv.DictWriter(f, fieldnames=header)
        writer.writeheader()
        writer.writerows(rows)


# 数値列の自動判定
# 列名を決め打ちせず、全ての値が数値に変換できる列だけを集計対象とみなす。
def numeric_columns(fieldnames, rows):
    """全行が数値に変換できる列名のリストを返す。"""
    numeric = []
    for col in fieldnames:
        values = [r[col] for r in rows if r.get(col, "") != ""]
        if not values:
            continue
        try:
            for v in values:
                float(v)
        except ValueError:
            continue
        numeric.append(col)
    return numeric


# 集計の共通処理
# sum / mean / count / min / max を1か所で計算し、各コマンドから使い回す。
def aggregate(values, how):
    """数値のリストに集計方法 how を適用した結果を返す。"""
    if how == "count":
        return len(values)
    if how == "sum":
        return sum(values)
    if how == "mean":
        return round(sum(values) / len(values), 2) if values else 0
    if how == "min":
        return min(values)
    if how == "max":
        return max(values)
    raise click.ClickException(f"未知の集計方法です: {how}")


def fmt(x):
    """整数とみなせる数値は小数点を省いて表示する。"""
    if isinstance(x, float) and x.is_integer():
        return str(int(x))
    return str(x)


# コマンド定義
@click.group()
def cli():
    """CSVを集計する小さなコマンドラインツール。"""


@cli.command()
@click.argument("path", type=click.Path())
def sample(path):
    """デモ用の売上CSVを PATH に書き出す。"""
    rows = [dict(zip(SAMPLE_HEADER, r)) for r in SAMPLE_ROWS]
    save_csv(path, SAMPLE_HEADER, rows)
    click.echo(f"サンプルCSVを書き出しました: {path}({len(rows)}件)")


@cli.command()
@click.argument("path", type=click.Path(exists=True))
@click.option("--output", "-o", type=click.Path(), default=None,
              help="集計結果をCSVに保存する")
def summary(path, output):
    """数値列を自動判定し、件数・合計・平均・最小・最大を集計する。"""
    fieldnames, rows = load_csv(path)
    cols = numeric_columns(fieldnames, rows)
    if not cols:
        raise click.ClickException("数値列が見つかりませんでした。")

    header = ["列", "件数", "合計", "平均", "最小", "最大"]
    result = []
    for col in cols:
        values = [float(r[col]) for r in rows if r.get(col, "") != ""]
        result.append(OrderedDict([
            ("列", col),
            ("件数", aggregate(values, "count")),
            ("合計", aggregate(values, "sum")),
            ("平均", aggregate(values, "mean")),
            ("最小", aggregate(values, "min")),
            ("最大", aggregate(values, "max")),
        ]))

    # 表として表示
    click.echo(" | ".join(header))
    for row in result:
        click.echo(" | ".join(fmt(row[h]) for h in header))

    # 任意でCSVへ保存
    if output:
        save_csv(output, header, [{k: fmt(v) for k, v in row.items()} for row in result])
        click.echo(f"集計結果を保存しました: {output}")


@cli.command()
@click.argument("path", type=click.Path(exists=True))
@click.option("--by", required=True, help="グループ化する列名")
@click.option("--target", required=True, help="集計する数値列名")
@click.option("--agg", "how",
              type=click.Choice(["sum", "mean", "count", "min", "max"]),
              default="sum", help="集計方法")
@click.option("--output", "-o", type=click.Path(), default=None,
              help="集計結果をCSVに保存する")
def groupby(path, by, target, how, output):
    """BY 列でグループ化し、TARGET 列を集計する。"""
    fieldnames, rows = load_csv(path)
    if by not in fieldnames:
        raise click.ClickException(f"列が見つかりません: {by}")
    if target not in fieldnames:
        raise click.ClickException(f"列が見つかりません: {target}")

    # グループごとに数値を集める
    groups = OrderedDict()
    for r in rows:
        val = r.get(target, "")
        if val == "":
            continue
        groups.setdefault(r[by], []).append(float(val))

    # グループ単位で集計して表示
    header = [by, how]
    result = []
    click.echo(" | ".join(header))
    for key, values in groups.items():
        agg_value = fmt(aggregate(values, how))
        result.append({by: key, how: agg_value})
        click.echo(f"{key} | {agg_value}")

    # 任意でCSVへ保存
    if output:
        save_csv(output, header, result)
        click.echo(f"集計結果を保存しました: {output}")


@cli.command(name="filter")
@click.argument("path", type=click.Path(exists=True))
@click.option("--column", "-c", required=True, help="絞り込む列名")
@click.option("--value", "-v", required=True, help="一致させる値")
@click.option("--output", "-o", type=click.Path(), default=None,
              help="絞り込み結果をCSVに保存する")
def filter_rows(path, column, value, output):
    """COLUMN 列が VALUE と一致する行だけを残す。"""
    fieldnames, rows = load_csv(path)
    if column not in fieldnames:
        raise click.ClickException(f"列が見つかりません: {column}")

    # 一致する行を抽出して表示
    matched = [r for r in rows if r.get(column, "") == value]
    click.echo(f"{len(matched)}件が一致しました(全{len(rows)}件中)")
    click.echo(" | ".join(fieldnames))
    for r in matched:
        click.echo(" | ".join(r.get(c, "") for c in fieldnames))

    # 任意でCSVへ保存
    if output:
        save_csv(output, fieldnames, matched)
        click.echo(f"絞り込み結果を保存しました: {output}")


def main():
    cli()


if __name__ == "__main__":
    main()

コード全文は上の折り畳みに入れてあるので、全部を上から読む必要はありません。ここでは特に重要な部分だけを抜き出して、何をしているのか順番に見ていきます。

utf-8-sigでCSVを読み込むload_csv

def load_csv(path):
    """CSVを読み込み、ヘッダーと行(dictのリスト)を返す。"""
    with open(path, newline="", encoding="utf-8-sig") as f:
        reader = csv.DictReader(f)
        rows = list(reader)
    return reader.fieldnames or [], rows

csv.DictReaderを使い、各行を見出し名をキーにした辞書として読み込みます。encodingにutf-8-sigを指定しているので、Excelが書き出したBOM付きCSVでも先頭の列名が壊れません。

全値がfloatに変換できる列を選ぶnumeric_columns

    for col in fieldnames:
        values = [r[col] for r in rows if r.get(col, "") != ""]
        if not values:
            continue
        try:
            for v in values:
                float(v)
        except ValueError:
            continue
        numeric.append(col)

列ごとに空でない値をすべてfloatに変換してみて、1つでも失敗すればその列を数値列から外します。列名を決め打ちしないので、金額でもpriceでも中身が数字なら自動で集計対象になります。

5種類の集計を1か所にまとめたaggregate

def aggregate(values, how):
    """数値のリストに集計方法 how を適用した結果を返す。"""
    if how == "count":
        return len(values)
    if how == "sum":
        return sum(values)
    if how == "mean":
        return round(sum(values) / len(values), 2) if values else 0

howという文字列で集計方法を切り替え、count・sum・meanなどを1つの関数で処理します。summaryとgroupbyの両方がこの関数を呼ぶため、集計ロジックの重複がありません。

未知の集計方法を弾くClickException

    if how == "min":
        return min(values)
    if how == "max":
        return max(values)
    raise click.ClickException(f"未知の集計方法です: {how}")

どの分岐にも当てはまらないhowが渡されたら、click.ClickExceptionで分かりやすいエラーを出します。Clickの例外を使うと、長いスタックトレースを見せずにメッセージだけを表示できます。

整数っぽい数値をきれいに見せるfmt

def fmt(x):
    """整数とみなせる数値は小数点を省いて表示する。"""
    if isinstance(x, float) and x.is_integer():
        return str(int(x))
    return str(x)

7600.0のように小数部が0の値は、intに直してから文字列にします。合計金額が7600と表示されるので、余計な小数点がつかず読みやすくなります。

@click.groupでコマンドをまとめるcli

@click.group()
def cli():
    """CSVを集計する小さなコマンドラインツール。"""

@click.group()を付けたcliが、4つのサブコマンドをぶら下げる親になります。この下に@cli.command()を付けた関数を並べるだけで、summaryやgroupbyが使えるようになります。

参考：

©Python公式ドキュメントcsvモジュール

If newline='' is not specified, newlines embedded inside quoted fields will not be interpreted correctly

実際にCLIツールを5回のコマンドで実行し、いずれも正常に(終了コード0で)動作しました。まずはsampleコマンドでデモ用の売上CSVsales.csvを書き出すところから始めます。

続いてsummary sales.csvを実行すると、数値列と判定された数量と金額について合計や平均などの統計量が表示されました。列名を指定しなくても数値列を拾えているのがポイントです。

groupby sales.csv --by カテゴリ --target 金額 --agg sumでは、カテゴリごとに金額を合計した表が得られました。filter sales.csv --column カテゴリ --value 家電では家電の行だけが抜き出されます。

最後にgroupbyへ--output by_category.csvを付け、集計結果をCSVファイルとして保存する実行も試しました。5回すべてが終了コード0で完了しています。

ここではcsvaggのようなCLIを動かすときに、初心者が一般に遭遇しやすいエラーをまとめます。特定の実行で必ず起きるものではなく、環境や入力次第で出やすい代表例です。

CSV集計CLIツールで注意したい点

動くコードを書けても、CSVやClickまわりには初心者がハマりやすい落とし穴がいくつかあります。事前に知っておくと、原因不明のエラーで悩む時間を減らせます。

csvaggは小さなツールですが、日々のちょっとした集計を自動化する場面で活躍します。表計算ソフトを開く手間を省けるのが利点です。

PythonのClickと標準ライブラリのcsvだけで、CSVを集計する小さなCLIcsvaggを作りました。@click.groupによるサブコマンド構成で、sample・summary・groupby・filterの4つを1つのツールにまとめています。

数値列の自動判定とaggregate関数の使い回しにより、少ないコードで柔軟な集計を実現できました。文字コードをutf-8-sigにそろえたことで、日本語CSVもExcelで扱いやすくなっています。

実際にCLIツールを5回のコマンドで実行し、すべて正常に(終了コード0で)動作しました。まずは手元のCSVをsummaryにかけるところから、追体験してみてください。

参考にした一次情報

1. ^ Click公式ドキュメント. https://click.palletsprojects.com/en/stable/, (参照26-07-19).
2. ^ Click: Commands and Groups(サブコマンドの作り方). https://click.palletsprojects.com/en/stable/commands/, (参照26-07-19).
3. ^ Python公式ドキュメント: csv — CSVファイルの読み書き. https://docs.python.org/ja/3/library/csv.html, (参照26-07-19).

※内容は執筆時点のものです。ライブラリやサイトの仕様は変わる可能性があるため、公式ドキュメントもあわせてご確認ください。

ITやプログラミングに関するコラム

• 
  【CSS】notで複数の件を除外する方法
• 
  【Git】remote設定を変更する方法
• 
  【VBA】コメントアウトを設定する方法
• 
  x86とx64の違いを分かりやすく解説
• 
  マークダウンで改行する方法
• 
  CapsLockキーを解除する方法
• 
  GitLabとGitHubの違いを解説
• 
  Linuxで環境変数を確認する方法
• 
  UbuntuのIPアドレスを確認する方法
• 
  パソコンのメモリの目安を用途別に選ぶ方法

ITやプログラミングに関するニュース

• 
  VercelがAI GatewayにSeedream 5.0 Proを追加、AI SDKのモデル指定で画像生成と編集が可能に
• 
  AWSがAmazon LocationのPlaces APIを強化、住所表記の指定と移動手段別の検索が可能に
• 
  VercelがトレースにTree・Waterfallビューを追加、ログ画面で処理の階層と所要時間を確認可能に
• 
  Googleがエージェント評価の再考を提唱、難易度を情報量で測るDiscovery Benchを解説
• 
  Google CloudがCloud Runサンドボックスを公開プレビューで提供、サービスヘルスは一般提供に
• 
  Google Cloud EMEAが英国金融の重要第三者に指定、イングランド銀行・PRA・FCAの直接監督下に
• 
  AWS DMS Schema ConversionがSQL Serverのオフライン変換に対応、ソースDBへ接続せずスキーマを変換可能に
• 
  EC2 G7インスタンスが米国東部（バージニア北部）で利用可能に、G6比でAI推論性能が最大4.6倍
• 
  SageMaker HyperPodが継続プロビジョニングでのAMIベース構成に対応、S3のスクリプト管理なしでSlurmクラスターを作成可能に
• 
  AWSがEMR on EKSでSparkトラブルシューティングエージェントに対応、失敗ジョブの原因分析を自然言語で依頼可能に