delhi09の勉強日記

技術トピック専用のブログです。自分用のメモ書きの投稿が多いです。あくまで「勉強日記」なので記事の内容は鵜呑みにしないでください。

Pythonでファイル読み込み時の例外の発生件数を集計する

概要

Pythonで10万行規模のテキストファイルを読み込んでいたところ、以下のような例外が発生して処理が途中で異常終了してしまった。

UnicodeDecodeError: 'utf-8' codec can't decode byte 0xae in position 5269: invalid start byte

ファイルにUTF-8にデコードできない文字が含まれているとのことである。

原因は理解できたのだが、1件発生したということは、他にもエラーになる行が存在する可能性が高いので、先に

  • 例外が発生する行の総件数
  • 発生するエラー内容の一覧

を把握しておきたいと思い、スクリプトを書いた。

環境

Python 3.7.3

スクリプト

以下、実際に書いたスクリプト

[check.py]

import os
import time

ERROR_FILE_NAME = "errors.txt"

process_start = time.time()
if os.path.exists(ERROR_FILE_NAME):
    os.remove(ERROR_FILE_NAME)

with open("target_file_name", "r", encoding="utf-8") as fr:
    error_count = 0
    line_number = 0
    while True:
        line_number += 1
        try:
            row = fr.readline()
        except Exception as e:
            error_count += 1
            with open(ERROR_FILE_NAME, "a", encoding="utf-8") as fw:
                fw.write(
                    "{:,}行目, エラーメッセージ: {} {}\n".format(
                        line_number, str(type(e)), str(e)
                    )
                )
        else:
            if not row:
                break

elapsed_time = time.time() - process_start
print(
    "エラー行数: {}/{:,}[行], 処理時間: {:.4f}[秒]".format(error_count, line_number, elapsed_time)
)

ポイント

Pythonでファイルを1行ずつ読み込む際には、一般的には

with open("target_file_name", "r", encoding="utf-8") as f:
    for row in f:
        # 処理

という書き方を使用するが、例外が発生するのはfor row in f:の部分なので、この書き方だと例外を捕まえられなくなってしまう。
(うまいやり方があるのかもしれないが、少なくとも私は分からなかった。)

従って、普段は使用しないreadline()関数を使用している。

実行例

$ python check.py
エラー行数: 42/174,313[行], 処理時間: 0.3259[秒]
$ head -3 errors.txt
155行目, エラーメッセージ: <class 'UnicodeDecodeError'> 'utf-8' codec can't decode byte 0xfc in position 6362: invalid start byte
5,262行目, エラーメッセージ: <class 'UnicodeDecodeError'> 'utf-8' codec can't decode byte 0xae in position 3764: invalid start byte
6,767行目, エラーメッセージ: <class 'UnicodeDecodeError'> 'utf-8' codec can't decode byte 0x96 in position 171: invalid start byte

備忘録

例外を発生させたくない場合は、open関数にerrorsという引数を指定することで対応可能。

詳細は以下の公式ドキュメントを参照。
docs.python.org