CSVを読めても、正しく読めたとは限らない

例えば timestamp,status を想定しているのに、途中から timestamp,status,note の3列になったら、後段の稼働率集計に誤ったデータが入る可能性があります。Pythonの csv.DictReader は列が多いと余りを restkey（既定値は None）に、少ないと不足値を restval（既定値は None）で表現します。エラーとして止めたい場合は、追加の検証が必要です。

この例では csv.reader(strict=True) を使い、ヘッダーの重複・列名と順序・各レコードの列数・空欄・件数上限を明示的に確認します。strict=True は不正なCSV構文に対するエラーであり、列数の一致を保証する設定ではありません。

確認用CSV

timestamp,status
2026-10-18T10:00:00+09:00,RUNNING
2026-10-18T10:05:00+09:00,STOPPED

この2レコードなら成功します。以下のように3列目が混入すると、行数・物理行番号を含むエラーで停止します。

timestamp,status
2026-10-18T10:00:00+09:00,RUNNING,unexpected

検証コード（標準ライブラリのみ）

csv_schema_check.py として保存します。utf-8-sig はUTF-8 BOM付きとBOMなしの両方を読み込めますが、CP932は対象外です。

"""CSV設備ログの列数・ヘッダー検証例（標準ライブラリのみ）。

CSVの構造だけを検査する。時刻順序や状態値の意味は別途検証が必要。
"""
import csv
from pathlib import Path


def validate_csv(path, expected=("timestamp", "status"), *,
                 max_bytes=1_048_576, max_rows=1000):
    """検証済みの辞書行を返す。不正入力は ValueError にする。"""
    path = Path(path)
    if path.stat().st_size > max_bytes:
        raise ValueError("CSVがサイズ上限を超えています")
    if not expected or len(set(expected)) != len(expected):
        raise ValueError("期待するヘッダーの指定が不正です")
    records = []
    try:
        with path.open("r", encoding="utf-8-sig", newline="") as f:
            reader = csv.reader(f, strict=True)
            header = next(reader, None)
            if not header:
                raise ValueError("ヘッダーがありません")
            if len(header) != len(set(header)):
                raise ValueError("ヘッダー名が重複しています")
            if header != list(expected):
                raise ValueError(f"列名または順序が違います: {header!r}")
            for row in reader:
                if len(row) != len(header):
                    raise ValueError(
                        f"CSV列数不一致（物理行 {reader.line_num}）:"
                        f" 期待{len(header)}列、実際{len(row)}列"
                    )
                if any(not cell.strip() for cell in row):
                    raise ValueError(f"空欄があります（物理行 {reader.line_num}）")
                if len(records) >= max_rows:
                    raise ValueError("CSVが行数上限を超えています")
                records.append(dict(zip(header, row)))
    except (csv.Error, UnicodeError) as exc:
        raise ValueError(f"CSVを解析できません: {exc}") from exc
    if not records:
        raise ValueError("データ行がありません")
    return records


if __name__ == "__main__":
    import sys
    if len(sys.argv) != 2:
        raise SystemExit("使い方: python3 csv_schema_check.py 設備ログ.csv")
    rows = validate_csv(sys.argv[1])
    print(f"構造チェック成功: {len(rows)}レコード")

実行例：python3 csv_schema_check.py sample.csv。正常時は 構造チェック成功: 2レコード と表示します。

列順の変更を許す運用なら header != list(expected) の判定を、列名集合の比較と辞書へのマッピングに置き換えます。ただし、列の意味を決めずに自動修復するのは避けてください。

テストと確認範囲

2026年10月18日にMac上のPython標準ライブラリ unittest で17件すべて成功しました。通常データ、UTF-8 BOM、クォートされたカンマと改行、CRLF、余分な列、欠けた列、重複ヘッダー、列順違い、空ファイル、ヘッダーのみ、空欄、不正な引用符、行数上限、ファイルサイズ上限、非UTF-8、誤った期待ヘッダーを検査しています。

テストコマンド：python3 -B -m unittest -v test_csv_schema_check.py。上限はサンプル固有の1MiB・1000レコードです。Path.stat() によるサイズ確認は読み込み前の値なので、別プロセスが同時にファイルを更新する状況で厳密なサイズ保証にはなりません。

残る検証と運用上の注意

• 日時の形式・タイムゾーン・時系列順序、状態値 RUNNING/STOPPED の妥当性はこのコードでは未検証です。
• csv.reader は空の物理行を通常スキップします。空行も異常扱いする場合は、別途検出ロジックを追加します。
• reader.line_num はCSVのレコード番号ではなく読み込んだ物理行数です。セル内に改行があれば一致しません。
• 巨大なフィールド、ファイルの同時更新、符号化方式の誤認、CSV数式注入、機密データの扱いには追加対策が必要です。
• 本番投入前に、実際のログ生成元のCSV仕様とテストデータを照合してください。