5日目のゴール
5日目のテーマは
「CSV処理を“バラバラのスクリプト”から、“ひとつのアプリ”としてまとめること」 です。
ここまでであなたは、
CSVを読む・書く
DictReader / DictWriter で列名ベースに扱う
集計・ソート・フィルタ
エラー行のスキップ・分離
までを、単発の関数として体験してきました。
今日はこれらをつなげて、
メニューで操作を選べる
同じCSVに対して、複数の処理を実行できる
「実務用のCSVユーティリティっぽいアプリ」を作る
ここを目指します。
アプリのイメージを先に言葉で決める
どんなことができるアプリにするか
まずは、今日作るアプリのイメージを言葉で固めます。
対象となるファイルは employees_dirty.csv。
メニューから、次のような処理を選べるようにします。
全件をざっと表示する
部署ごとの給与合計・平均を表示する
30歳以上の社員だけを別CSVに書き出す
エラー行（年齢・給与がおかしい行）だけを別CSVに書き出す
「実務でよくやる“ちょっとしたCSV処理”を、ひとつのアプリにまとめた」
そんなイメージです。
共通の読み込み関数を用意する
何度も同じファイルを読むなら、関数にする
まずは、どの機能からも使える
「安全な読み込み関数」を用意します。
import csv
def load_employees(filename):
try:
with open(filename, "r", encoding="utf-8", newline="") as f:
reader = csv.DictReader(f)
rows = list(reader)
except FileNotFoundError:
print(f"ファイルが見つかりません: {filename}")
return []
except PermissionError:
print(f"ファイルにアクセスできません（権限エラー）: {filename}")
return []
except OSError as e:
print("ファイルの読み込み中にエラーが発生しました。")
print("詳細:", e)
return []
print(f"{filename} から {len(rows)} 行読み込みました。")
return rows
ここで深掘りしたいポイントは二つです。
一つ目は、「ファイルを開くところ」を try / except で囲んでいること。
二つ目は、エラー時には空リストを返し、呼び出し側で「データなし」として扱えるようにしていること。
この関数を使うことで、
どの機能も「まず load_employees を呼ぶ」という共通の入り口を持てます。
全件をざっと表示する機能
「まず中身を見たい」というニーズに応える
最初の機能は、とてもシンプルです。
読み込んだ行を、そのまま人間が読める形で表示します。
def show_all_employees(rows):
if not rows:
print("データがありません。")
return
print("=== 全社員一覧（先頭10件まで） ===")
for i, row in enumerate(rows):
if i >= 10:
print("...（続きはCSVファイルを直接確認してください）")
break
print(f"ID={row['id']}, 名前={row['name']}, 年齢={row['age']}, 部署={row['department']}, 給与={row['salary']}")
ここでのポイントは、
行数が多いことを想定して「先頭10件だけ」にしていること。
空リストのときは、何もせずメッセージだけ出して終わること。
「まずざっと中身を確認する」というのは、
実務で一番最初にやる動きなので、
アプリの機能として持っておくと便利です。
部署ごとの給与合計・平均を表示する機能
3日目でやった集計を“アプリの機能”にする
次に、部署ごとの給与合計・平均を出す機能を
関数として切り出します。
def show_salary_by_department(rows):
if not rows:
print("データがありません。")
return
stats = {}
skipped = 0
for row in rows:
dept = row["department"]
salary_text = row["salary"]
if dept == "" or salary_text == "":
skipped += 1
continue
try:
salary = int(salary_text)
except ValueError:
skipped += 1
continue
if dept not in stats:
stats[dept] = {"total_salary": 0, "count": 0}
stats[dept]["total_salary"] += salary
stats[dept]["count"] += 1
if not stats:
print("有効な給与データがありませんでした。")
return
print("=== 部署ごとの給与集計 ===")
for dept, info in stats.items():
total = info["total_salary"]
count = info["count"]
average = total / count
print(f"[{dept}] 社員数: {count} 人, 合計: {total} 円, 平均: {average:.1f} 円")
print(f"スキップした行数: {skipped}")
ここで深掘りしたいのは三つです。
部署や給与が空欄の行は、最初に if で弾いていること。
給与が数値でない行は、try / except で弾いていること。
弾いた行数を skipped として数え、最後に報告していること。
これで、「集計しつつ、どれくらいデータが汚れていたか」も分かるようになります。
30歳以上の社員だけを別CSVに書き出す機能
フィルタ＋書き出しをひとつの機能にまとめる
次は、「30歳以上だけを抽出して別ファイルに保存する」機能です。
def export_over_30(rows, output_file):
if not rows:
print("データがありません。")
return
import csv
filtered = []
skipped = 0
for row in rows:
age_text = row["age"]
if age_text == "":
skipped += 1
continue
try:
age = int(age_text)
except ValueError:
skipped += 1
continue
if age >= 30:
filtered.append(row)
if not filtered:
print("30歳以上のデータがありませんでした。")
print(f"スキップした行数: {skipped}")
return
fieldnames = rows[0].keys()
with open(output_file, "w", encoding="utf-8", newline="") as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
for row in filtered:
writer.writerow(row)
print(f"{output_file} に 30歳以上のデータを書き出しました。")
print(f"対象件数: {len(filtered)} 件, スキップした行数: {skipped}")
ここでの重要ポイントは、
「フィルタ」と「書き出し」をひとつの関数にまとめていること。
年齢が空欄・数値でない行は、スキップしていること。
書き出す列構成は、元の rows の先頭行のキーから取っていること。
これで、「条件で絞って別CSVに出す」という
実務でよくある動きが、アプリのメニューから選べるようになります。
エラー行だけを別CSVに書き出す機能
「正常行」と「エラー行」を分けるユーティリティとして使える
4日目でやった「エラー行の分離」を、
アプリの機能として整理します。
ここでは、「年齢と給与が両方とも数値でない行」を
エラー行とみなします。
def export_error_rows(rows, output_file):
if not rows:
print("データがありません。")
return
import csv
error_rows = []
for row in rows:
age_text = row["age"]
salary_text = row["salary"]
try:
int(age_text)
int(salary_text)
except (ValueError, TypeError):
error_rows.append(row)
if not error_rows:
print("エラー行はありませんでした。")
return
fieldnames = rows[0].keys()
with open(output_file, "w", encoding="utf-8", newline="") as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
for row in error_rows:
writer.writerow(row)
print(f"{output_file} にエラー行を書き出しました。件数: {len(error_rows)} 件")
ここでのポイントは、
「何をエラーとみなすか」を関数の中で定義していること。
エラー行だけを別CSVに残すことで、後から人間が確認できるようにしていること。
この機能は、
「集計前にデータをきれいにしたいとき」や
「データ提供元に“ここがおかしいです”と伝えたいとき」に役立ちます。
メニューとメインループを作る
これまでの関数を“アプリ”としてまとめる
ここまで作った機能を、
メニューから選べるようにします。
DATA_FILE = "employees_dirty.csv"
def show_menu():
print("==========")
print("CSVファイル読み書きアプリ（5日目）")
print("1: 全社員をざっと表示する")
print("2: 部署ごとの給与集計を表示する")
print("3: 30歳以上の社員を別CSVに書き出す")
print("4: エラー行を別CSVに書き出す")
print("0: 終了")
print("==========")
def main():
rows = load_employees(DATA_FILE)
if not rows:
print("処理を終了します。")
return
while True:
show_menu()
choice = input("番号を選んでください: ").strip()
if choice == "1":
show_all_employees(rows)
elif choice == "2":
show_salary_by_department(rows)
elif choice == "3":
export_over_30(rows, "employees_over30.csv")
elif choice == "4":
export_error_rows(rows, "employees_error.csv")
elif choice == "0":
print("アプリを終了します。")
break
else:
print("0〜4 の番号を入力してください。")
if __name__ == "__main__":
main()
ここで深掘りしたいのは三つです。
アプリ起動時に一度だけ load_employees を呼び、rows を読み込んでいること。
メニューから選んだ番号に応じて、対応する関数を呼び分けていること。
「アプリとしての入り口」が main() にまとまっていること。
これで、
「CSVを対象にした小さな実務ツール」が
ひとつの Python ファイルとして完成します。
5日目のまとめ 今日つかんでほしい感覚
今日の本質は、これです。
同じCSVに対して複数の処理をするなら、「共通の読み込み関数」を用意する。
集計・フィルタ・エラー分離などの処理は、それぞれを関数として独立させる。
メニューとメインループを作ると、「スクリプト」ではなく「アプリ」になる。
「どの機能がどの関数に対応しているか」が、自分の頭の中で整理されていることが大事。
ここまで来たあなたは、
csvモジュールを「ただ知っている」段階を超えて、
“CSVを扱う小さな実務アプリを自分で設計できる人” に近づいています。
6日目・7日目では、
このアプリを少しずつリファクタリングしたり、
設定ファイル化・クラス化などにも触れていきます。
