Python | 1 日 120 分 × 7 日アプリ学習：CSVファイル読み書きアプリ（中級編）

Web APP Python
スポンサーリンク

4日目のゴール

4日目のテーマは
「CSVを“きれいなデータだけ”前提で扱うのをやめて、エラー行や欠損値があっても、落ちずに処理できるようになること」 です。

1〜3日目で、あなたはすでに

CSVを読む・書く
DictReader / DictWriter で列名ベースで扱う
集計・ソート・フィルタで“実務っぽい処理”を書く

ところまで来ています。

4日目ではここに、

ファイル読み込み時のエラー処理
壊れた行・おかしな値を「スキップする」発想
「スキップした件数」をメッセージとして出す

という、「現場で本当に必要になる CSV 処理の現実対応」を足していきます。

まずは前提：実務のCSVはきれいじゃない

「全部ちゃんと入っている」は幻想だと思っておく

教科書に出てくるCSVは、だいたいこうです。

id,name,age,department,salary
1,Taro,25,Sales,300000
2,Hanako,30,HR,320000
3,Ken,22,Sales,250000

でも、実務で渡されるCSVは、こんな感じになりがちです。

id,name,age,department,salary
1,Taro,25,Sales,300000
2,Hanako,,HR,320000
3,Ken,abc,Sales,250000
4,Naomi,35,,400000
5,Shin,28,Engineering,

年齢が空欄
年齢が「abc」みたいな文字列
部署が空欄
給与が空欄

こういう「おかしな行」が混ざっているのが普通です。

今日やるのは、
「こういう行があっても、アプリを落とさずに処理を続ける」
という考え方です。

数値変換でエラーが出るパターンを体験する

まずはあえて「落ちるコード」を書いてみる

さっきのような「汚れた employees_dirty.csv」を読み込んで、
年齢と給与の平均を出すコードを書いてみます。

import csv

def calc_average_from_dirty():
    with open("employees_dirty.csv", "r", encoding="utf-8", newline="") as f:
        reader = csv.DictReader(f)

        total_age = 0
        total_salary = 0
        count = 0

        for row in reader:
            age = int(row["age"])
            salary = int(row["salary"])

            total_age += age
            total_salary += salary
            count += 1

    print("件数:", count)
    print("平均年齢:", total_age / count)
    print("平均給与:", total_salary / count)
Python

このまま実行すると、
int(row["age"]) のところで ValueError が出ます。

理由はシンプルで、
""（空文字）や "abc" は int() に変換できないからです。

ここで大事なのは、
「現実のCSVは、int() できない値が普通に混ざる」
という感覚を持つことです。

try / except で「変換できない行」をスキップする

「落とさない」ことを最優先にする

次に、「変換できない行はスキップする」ように書き換えます。

import csv

def calc_average_from_dirty_safe():
    with open("employees_dirty.csv", "r", encoding="utf-8", newline="") as f:
        reader = csv.DictReader(f)

        total_age = 0
        total_salary = 0
        count = 0
        skipped_rows = 0

        for row in reader:
            try:
                age = int(row["age"])
                salary = int(row["salary"])
            except ValueError:
                skipped_rows += 1
                print("変換できない行をスキップしました:", row)
                continue

            total_age += age
            total_salary += salary
            count += 1

    if count == 0:
        print("有効なデータがありませんでした。")
        return

    print("有効な件数:", count)
    print("スキップした行数:", skipped_rows)
    print("平均年齢:", total_age / count)
    print("平均給与:", total_salary / count)
Python

ここで深掘りしたいポイントは四つです。

try: ... except ValueError: で「数値変換に失敗した行だけを捕まえている」こと。
失敗した行は continue で「この1行だけ飛ばして、次の行に進んでいる」こと。
スキップした行数を skipped_rows で数えて、最後に報告していること。
有効な件数が 0 のときは、平均を出さずにメッセージを出していること。

つまり、

「全部きれいに処理する」のではなく
「処理できる行だけを使い、できなかった行は数えておく」

という現実的なスタンスに変わっています。

必須項目が空欄の行をスキップする

「値がない」パターンもちゃんと見る

次は、「空欄」のパターンです。

例えば、部署や給与が空欄の行は
集計対象から外したい、というケース。

import csv

def calc_average_salary_with_required_fields():
    with open("employees_dirty.csv", "r", encoding="utf-8", newline="") as f:
        reader = csv.DictReader(f)

        total_salary = 0
        count = 0
        skipped_rows = 0

        for row in reader:
            if row["salary"] == "" or row["department"] == "":
                skipped_rows += 1
                print("必須項目が空のためスキップ:", row)
                continue

            try:
                salary = int(row["salary"])
            except ValueError:
                skipped_rows += 1
                print("給与が数値でないためスキップ:", row)
                continue

            total_salary += salary
            count += 1

    if count == 0:
        print("有効なデータがありませんでした。")
        return

    print("有効な件数:", count)
    print("スキップした行数:", skipped_rows)
    print("平均給与:", total_salary / count)
Python

ここでの重要ポイントは、

「空欄チェック」と「数値変換エラー」を分けて扱っていること。
空欄のときは if で弾き、変換エラーは try / except で弾いていること。
どちらも「スキップした理由」をメッセージとして出していること。

この「スキップの理由を分けておく」習慣は、
後でデータ提供元にフィードバックするときにも役立ちます。

「エラー行だけを別CSVに書き出す」ミニアプリ

スキップした行を“捨てずに残す”という発想

実務では、
「おかしな行をスキップするだけでなく、別ファイルに残しておきたい」
というニーズもよくあります。

例えば、

employees_dirty.csv を読み込む
年齢と給与が両方とも数値の行だけを「clean_employees.csv」に書き出す
それ以外の行は「error_employees.csv」に書き出す

という処理を考えてみます。

import csv

def split_clean_and_error_rows(input_file, clean_file, error_file):
    with open(input_file, "r", encoding="utf-8", newline="") as f_in:
        reader = csv.DictReader(f_in)
        fieldnames = reader.fieldnames

        clean_rows = []
        error_rows = []

        for row in reader:
            try:
                age = int(row["age"])
                salary = int(row["salary"])
            except (ValueError, TypeError):
                error_rows.append(row)
                continue

            row["age"] = age
            row["salary"] = salary
            clean_rows.append(row)

    with open(clean_file, "w", encoding="utf-8", newline="") as f_clean:
        writer_clean = csv.DictWriter(f_clean, fieldnames=fieldnames)
        writer_clean.writeheader()
        for row in clean_rows:
            row_to_write = {
                "id": row["id"],
                "name": row["name"],
                "age": str(row["age"]),
                "department": row["department"],
                "salary": str(row["salary"]),
            }
            writer_clean.writerow(row_to_write)

    with open(error_file, "w", encoding="utf-8", newline="") as f_error:
        writer_error = csv.DictWriter(f_error, fieldnames=fieldnames)
        writer_error.writeheader()
        for row in error_rows:
            writer_error.writerow(row)

    print(f"正常行: {len(clean_rows)} 件 → {clean_file}")
    print(f"エラー行: {len(error_rows)} 件 → {error_file}")
Python

ここで深掘りしたいポイントは、かなり実務的です。

reader.fieldnames で「元のCSVのヘッダー（列名）」をそのまま使っていること。
正常行とエラー行を、それぞれ別のリストにためていること。
正常行は数値に変換してから、書き出すときに文字列に戻していること。
エラー行は「そのまま」別CSVに書き出していること。

これで、

「きれいなデータだけを使って集計しつつ、
おかしな行は後で確認できるように残しておく」

という、かなり“現場っぽい”動きになります。

ファイル自体のエラーもちゃんと扱う

ファイルがない・壊れているときに落とさない

ここまで「行レベル」のエラーを見てきましたが、
「ファイルレベル」のエラーもあります。

ファイルが存在しない
権限がなくて開けない
中身がJSONやExcelで、CSVじゃない

こういうときに、
アプリがドカンと落ちるのは避けたいですよね。

import csv

def safe_read_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
Python

ここでの重要ポイントは、

try: ... except ... で「ファイルを開くところ」全体を囲んでいること。
エラーの種類ごとにメッセージを変えていること。
エラー時には空のリスト [] を返して、呼び出し側で「データなし」として扱えるようにしていること。

これで、呼び出し側はこう書けます。

rows = safe_read_employees("employees_dirty.csv")
if not rows:
    print("処理を中止します。")
    return
Python

「ファイルがないなら、ないなりに止まる」
という、落ち着いた動きになります。

4日目のまとめ 今日つかんでほしい感覚

今日の本質は、これです。

実務のCSVは「きれいじゃない」のが普通だと最初から思っておく。
数値変換は必ず try / except で囲み、変換できない行はスキップする。
必須項目が空欄の行は、if で弾いて「スキップした理由」をメッセージに残す。
正常行とエラー行を分けて、エラー行は別CSVに書き出すと“現場で使える”形になる。
ファイル自体のエラー（存在しない・権限なし）も try / except で受け止めて、アプリを落とさない。

1〜3日目で「きれいなCSVを自由に料理できる力」がつき、
4日目で「汚れたCSVでも落ちずに処理できる力」が乗りました。

この二つがそろったとき、
あなたのCSV処理は、もう立派な“実務レベル”です。

前へ次へ
1234567
スポンサーリンク
タイトルとURLをコピーしました