11. ファイル入出力と標準ライブラリ
テーマ: 売上CSVの生成・読み込み・集計・書き出し
学習点: csv モジュール, with 文, collections(defaultdict/Counter), pathlib, datetime, sorted の key 引数
依存: 標準ライブラリのみ ／ 難易度: 中級
実行方法
uv run 11_csv_aggregate.py
スクリプト冒頭の PEP 723 メタデータ(# /// script)により、必要なライブラリは
uv が自動的に仮想環境へ導入します。事前の pip install は不要です。
解説
何をするプログラムか
実務のデータ分析では、会計システムや POS から出力された CSV ファイルを読み込み、集計し、結果をまた CSV で共有する、という流れが最も基本的な作業です。このスクリプトは、その一連の流れ（生成 → 読み込み → 加工 → 集計 → 書き出し）を pandas などの外部ライブラリを一切使わず、Python の標準ライブラリだけで実装しています。
まず疑似的な売上データ 400 件を out_11/sales.csv に書き出し、それを読み直して地域別・商品別・月次の売上を集計し、最後に地域別サマリを summary_by_region.csv として出力します。標準ライブラリだけでどこまでできるかを知っておくと、pandas が使えない環境や小規模な処理で役立ちます。
コードの読みどころ
OUT = Path("out_11")とOUT.mkdir(exist_ok=True)はpathlibの典型的な使い方です。OUT / "sales.csv"のように/演算子でパスを連結でき、src.stat().st_sizeでファイルサイズも取得できます。generate()ではwith path.open("w", newline="", encoding="utf-8")としてファイルを開きます。with文によりブロックを抜けると自動でクローズされ、csvモジュールでは改行の二重変換を防ぐためnewline=""の指定が必須です。- 日付は
date(2025, 1, 1) + timedelta(days=random.randrange(365))で生成し、d.isoformat()で2025-04-01形式の文字列にします。読み込み後はr["date"][:7]の文字列スライスだけで「月」を取り出せます。 - 読み込みは
csv.DictReaderを使い、各行をr["unit_price"]のように列名でアクセスできる辞書として受け取ります。CSV の値はすべて文字列なので、int(r["unit_price"]) * int(r["qty"])と明示的に変換して金額列を追加しています。 - 集計は
by_region = defaultdict(int)が要です。キーが未登録でも初期値 0 から+=できるため、「キーの存在確認 → 初期化 → 加算」の 3 手順が 1 行になります。件数のカウントにはCounter(...).most_common(3)を使っています。 - 表示順は
sorted(by_region.items(), key=lambda kv: -kv[1])で金額の降順に制御します。sortedのkey引数に「何で並べるか」を関数として渡すのがポイントです。 - 書き出しは
csv.DictWriterにfieldnamesを与え、writeheader()の後に辞書を 1 行ずつwriterow()しています。
実行結果の見方
先頭行に生成された CSV のサイズ（17,773 bytes）、続いて総売上 95,903,700 円が表示されます。地域別売上では福岡が 28.8% で首位、シェアに比例した棒（█）で構成比が一目で分かります。商品別では単価 12.8 万円のノートPC が約 6,323 万円と、総売上の 3 分の 2 を占めることが読み取れます。
末尾の [出現回数] は金額ではなく取引件数ベースの上位 3 地域で、Counter の結果がタプルのリストで表示されます。最後に書き出した summary_by_region.csv の中身をそのまま表示しており、シェア（share 列）の合計が 1 になることを確認できます。
ソースコード
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
"""11: ファイル入出力と標準ライブラリ -------------------------------------
テーマ: 売上CSVの生成・読み込み・集計・書き出し
学習点: csv モジュール, with 文, collections(defaultdict/Counter),
pathlib, datetime, sorted の key 引数
"""
import csv
import random
from collections import Counter, defaultdict
from datetime import date, timedelta
from pathlib import Path
random.seed(7)
OUT = Path("out_11")
OUT.mkdir(exist_ok=True)
PRODUCTS = [("P001", "ノートPC", 128000), ("P002", "モニタ", 32000),
("P003", "キーボード", 8500), ("P004", "マウス", 4200),
("P005", "ドック", 21000)]
REGIONS = ["東京", "大阪", "名古屋", "福岡", "札幌"]
def generate(path: Path, n: int = 400) -> None:
"""疑似売上データを CSV に書き出す。"""
start = date(2025, 1, 1)
with path.open("w", newline="", encoding="utf-8") as f:
w = csv.writer(f)
w.writerow(["date", "region", "product_id", "product", "unit_price", "qty"])
for _ in range(n):
d = start + timedelta(days=random.randrange(365))
pid, name, price = random.choice(PRODUCTS)
w.writerow([d.isoformat(), random.choice(REGIONS), pid, name,
price, random.randint(1, 12)])
def main() -> None:
src = OUT / "sales.csv"
generate(src)
print(f"生成: {src} ({src.stat().st_size:,} bytes)")
# DictReader は1行を辞書として返す
with src.open(encoding="utf-8") as f:
rows = list(csv.DictReader(f))
for r in rows:
r["amount"] = int(r["unit_price"]) * int(r["qty"])
r["month"] = r["date"][:7]
print(f"レコード数: {len(rows):,} / 総売上: {sum(r['amount'] for r in rows):,}円")
by_region = defaultdict(int)
by_product = defaultdict(int)
by_month = defaultdict(int)
for r in rows:
by_region[r["region"]] += r["amount"]
by_product[r["product"]] += r["amount"]
by_month[r["month"]] += r["amount"]
total = sum(by_region.values())
print("\n[地域別売上]")
for k, v in sorted(by_region.items(), key=lambda kv: -kv[1]):
print(f" {k:<5}{v:>13,}円 {v/total:>7.1%} {'█' * int(v/total*60)}")
print("\n[商品別売上 上位3]")
for k, v in sorted(by_product.items(), key=lambda kv: -kv[1])[:3]:
print(f" {k:<8}{v:>13,}円")
print("\n[月次推移]")
for m in sorted(by_month):
v = by_month[m]
print(f" {m} {v:>12,}円 {'*' * int(v / 400000)}")
print(f"\n[出現回数] {Counter(r['region'] for r in rows).most_common(3)}")
# 集計結果を新しい CSV に書き出す
dst = OUT / "summary_by_region.csv"
with dst.open("w", newline="", encoding="utf-8") as f:
w = csv.DictWriter(f, fieldnames=["region", "amount", "share"])
w.writeheader()
for k, v in sorted(by_region.items(), key=lambda kv: -kv[1]):
w.writerow({"region": k, "amount": v, "share": f"{v/total:.4f}"})
print(f"\n書き出し: {dst}")
print(dst.read_text(encoding="utf-8"))
if __name__ == "__main__":
main()
実行結果
生成: out_11/sales.csv (17,773 bytes)
レコード数: 400 / 総売上: 95,903,700円
[地域別売上]
福岡 27,637,900円 28.8% █████████████████
東京 23,319,000円 24.3% ██████████████
札幌 17,294,500円 18.0% ██████████
大阪 14,580,600円 15.2% █████████
名古屋 13,071,700円 13.6% ████████
[商品別売上 上位3]
ノートPC 63,232,000円
モニタ 16,832,000円
ドック 8,883,000円
[月次推移]
2025-01 9,868,700円 ************************
2025-02 7,773,300円 *******************
2025-03 4,526,400円 ***********
2025-04 12,108,200円 ******************************
2025-05 9,359,400円 ***********************
2025-06 5,461,600円 *************
2025-07 9,943,500円 ************************
2025-08 6,817,500円 *****************
2025-09 6,196,800円 ***************
2025-10 5,680,000円 **************
2025-11 11,466,800円 ****************************
2025-12 6,701,500円 ****************
[出現回数] [('福岡', 94), ('大阪', 83), ('東京', 81)]
書き出し: out_11/summary_by_region.csv
region,amount,share
福岡,27637900,0.2882
東京,23319000,0.2432
札幌,17294500,0.1803
大阪,14580600,0.1520
名古屋,13071700,0.1363