第158回では、CSVの列構成をv1からv2へ移行する処理を扱いました。しかし、その前段で文字コードを正しく扱えなければ、移行処理には進めません。
今回は、受け取ったCSVをバイト列で読み、まずUTF-8として厳密に検査します。UTF-8でなかった場合は、明示的に許可したときだけCP932として読みます。文字コードを自動推測せず、受け入れる形式を業務ルールとして限定する設計です。
今回の入出力契約
| 項目 | 仕様 |
|---|---|
| 既定の入力 | BOMなしUTF-8、または先頭にUTF-8 BOMが1個だけあるUTF-8 |
| 任意の入力 | --allow-cp932を指定した場合のみCP932 |
| 対象外 | UTF-16、UTF-32、およびUTF-8 BOMが2個以上連続する入力 |
| 出力 | BOMなしUTF-8 |
| CSV条件 | 1行目に少なくとも1つ値のあるヘッダーがあり、すべてのレコードがヘッダーと同じ列数であること |
| 上書き | 既存ファイルへの上書きは禁止 |
| 変換成功時 | 標準出力へ1行のJSON、終了コード0 |
| スクリプト内で検出したエラー | 標準エラー出力へ1行のJSON、終了コード2～5 |
CP932は自動判定するのではなく、UTF-8でなかった場合の代替候補です。入力元がCP932を出力すると確認できている業務だけで有効にします。
UTF-8 BOMは先頭に1個だけ許可します。複数のBOMを成功扱いにすると、utf-8-sigで除去されなかったBOMが出力へ残るため、この実装では文字コードエラーとして拒否します。
処理の順序
- 入力を文字列ではなくバイト列として読む
- UTF-16またはUTF-32のBOMがあれば対象外として停止する
- UTF-8 BOMが2個以上連続していれば停止する
- 単一のUTF-8 BOM、BOMなしUTF-8、許可時のCP932の順で厳密にデコードする
- CSVとして解析し、ヘッダーと列数を確認する
- BOMなしUTF-8へ変換し、新しいファイルとして保存する
- 入力文字コード、列数、データ行数をJSONで報告する
CSVの検査にはcsv.readerを使いますが、出力をcsv.writerでは再生成しません。引用符や改行コードを組み直さず、デコードした文字列をUTF-8へ変換します。単一のBOMが付いたUTF-8入力では、そのBOMだけを除いて内容を維持します。
標準ライブラリだけで実装する
次のコードをnormalize_csv_encoding.pyとして保存します。型注釈にはPython 3.9以降の記法を使用しています。
import argparse
import codecs
import csv
import io
import json
import sys
from pathlib import Path
EXIT_OK = 0
EXIT_INPUT = 2
EXIT_ENCODING = 3
EXIT_CSV = 4
EXIT_OUTPUT = 5
class EncodingRejected(Exception):
pass
class CsvRejected(Exception):
pass
class OutputRejected(Exception):
pass
def decode_input(data: bytes, allow_cp932: bool) -> tuple[str, str]:
unsupported_boms = (
codecs.BOM_UTF32_LE,
codecs.BOM_UTF32_BE,
codecs.BOM_UTF16_LE,
codecs.BOM_UTF16_BE,
)
if data.startswith(unsupported_boms):
raise EncodingRejected(
'UTF-16/32は対象外です。送信元でUTF-8またはCP932として書き出してください'
)
if data.startswith(codecs.BOM_UTF8 * 2):
raise EncodingRejected('UTF-8 BOMが複数連続しているため受け入れません')
if data.startswith(codecs.BOM_UTF8):
try:
return data.decode('utf-8-sig', errors='strict'), 'utf-8-sig'
except UnicodeDecodeError as exc:
raise EncodingRejected(
'UTF-8 BOMがありますが、本文をUTF-8としてデコードできません'
) from exc
try:
return data.decode('utf-8', errors='strict'), 'utf-8'
except UnicodeDecodeError as utf8_error:
if not allow_cp932:
raise EncodingRejected(
'入力はUTF-8ではありません。CP932を受け入れる場合だけ '
'--allow-cp932 を指定してください'
) from utf8_error
try:
return data.decode('cp932', errors='strict'), 'cp932'
except UnicodeDecodeError as cp932_error:
raise EncodingRejected(
'入力をUTF-8またはCP932として厳密にデコードできません'
) from cp932_error
def inspect_csv(text: str) -> tuple[int, int]:
if '\x00' in text:
raise CsvRejected('NUL文字を含むためCSVとして受け入れません')
reader = csv.reader(io.StringIO(text, newline=''), strict=True)
try:
header = next(reader)
except StopIteration as exc:
raise CsvRejected('CSVが空です') from exc
except csv.Error as exc:
raise CsvRejected('ヘッダーをCSVとして解析できません') from exc
if not header or all(cell == '' for cell in header):
raise CsvRejected('有効なヘッダーがありません')
expected_columns = len(header)
data_rows = 0
try:
for record_number, row in enumerate(reader, start=2):
if len(row) != expected_columns:
raise CsvRejected(
f'レコード{record_number}の列数は{len(row)}です。'
f'ヘッダーの{expected_columns}列と一致しません'
)
data_rows += 1
except csv.Error as exc:
raise CsvRejected(
f'CSV構文を解析できません（物理行{reader.line_num}付近）'
) from exc
return expected_columns, data_rows
def write_new_file(path: Path, payload: bytes) -> None:
created = False
try:
handle = path.open('xb')
created = True
with handle:
written = handle.write(payload)
if written != len(payload):
raise OSError('short write')
except FileExistsError as exc:
raise OutputRejected(f'出力先が既に存在します: {path}') from exc
except OSError as exc:
if created:
try:
path.unlink()
except OSError:
pass
raise OutputRejected(f'出力先へ書き込めません: {path}') from exc
def emit_error(code: int, message: str) -> int:
report = {'status': 'error', 'message': message}
print(
json.dumps(report, ensure_ascii=False, separators=(',', ':')),
file=sys.stderr,
)
return code
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description='CSVを検査し、BOMなしUTF-8へ変換します'
)
parser.add_argument('input', type=Path, help='入力CSV')
parser.add_argument('output', type=Path, help='新規作成する出力CSV')
parser.add_argument(
'--allow-cp932',
action='store_true',
help='UTF-8でない入力をCP932として試します',
)
return parser.parse_args()
def main() -> int:
args = parse_args()
try:
data = args.input.read_bytes()
except OSError:
return emit_error(EXIT_INPUT, f'入力を読み込めません: {args.input}')
try:
text, source_encoding = decode_input(data, args.allow_cp932)
except EncodingRejected as exc:
return emit_error(EXIT_ENCODING, str(exc))
try:
columns, data_rows = inspect_csv(text)
except CsvRejected as exc:
return emit_error(EXIT_CSV, str(exc))
try:
write_new_file(args.output, text.encode('utf-8'))
except OutputRejected as exc:
return emit_error(EXIT_OUTPUT, str(exc))
report = {
'status': 'ok',
'source_encoding': source_encoding,
'output_encoding': 'utf-8',
'columns': columns,
'data_rows': data_rows,
'input': str(args.input),
'output': str(args.output),
}
print(json.dumps(report, ensure_ascii=False, separators=(',', ':')))
return EXIT_OK
if __name__ == '__main__':
raise SystemExit(main())
再現手順
1．CP932の入力例を作る
customers_utf8.csvが存在しない作業ディレクトリで、次のコマンドを実行します。
python -c "from pathlib import Path; Path('customers_cp932.csv').write_bytes('顧客ID,氏名\r\nC001,山田太郎\r\n'.encode('cp932'))"2．許可せずに実行する
python normalize_csv_encoding.py customers_cp932.csv customers_utf8.csv
echo $?CP932を許可していないため、標準エラー出力と終了コードは次のようになります。出力CSVは作られません。
{"status":"error","message":"入力はUTF-8ではありません。CP932を受け入れる場合だけ --allow-cp932 を指定してください"}
33．入力元を確認したうえでCP932を許可する
python normalize_csv_encoding.py customers_cp932.csv customers_utf8.csv --allow-cp932
echo $?正常時は標準出力へ1行のJSONが出ます。
{"status":"ok","source_encoding":"cp932","output_encoding":"utf-8","columns":2,"data_rows":1,"input":"customers_cp932.csv","output":"customers_utf8.csv"}
0出力にBOMがなく、UTF-8として読めることも確認します。
python -c "from pathlib import Path; b=Path('customers_utf8.csv').read_bytes(); print(b.startswith(bytes((239, 187, 191))), b.decode('utf-8').splitlines()[0])"False 顧客ID,氏名4．UTF-8 BOMが複数連続する入力を拒否する
codecs.BOM_UTF8を2個付けたCSVを作り、未使用の出力先を指定します。
python -c "from pathlib import Path; import codecs; Path('double_bom.csv').write_bytes(codecs.BOM_UTF8 * 2 + b'a,b\n1,2\n')"
python normalize_csv_encoding.py double_bom.csv double_bom_output.csv
echo $?この入力は文字コードエラーになり、出力CSVは作られません。
{"status":"error","message":"UTF-8 BOMが複数連続しているため受け入れません"}
35．列数が異なるCSVを拒否する
python -c "from pathlib import Path; Path('broken.csv').write_bytes(b'id,name\n1\n')"
python normalize_csv_encoding.py broken.csv unused.csv
echo $?{"status":"error","message":"レコード2の列数は1です。ヘッダーの2列と一致しません"}
4検査に失敗したため、unused.csvは作られません。
6．既存の出力ファイルを上書きしないことを確認する
手順3で作成したcustomers_utf8.csvを出力先にして、同じ処理をもう一度実行します。
python normalize_csv_encoding.py customers_cp932.csv customers_utf8.csv --allow-cp932
echo $?{"status":"error","message":"出力先が既に存在します: customers_utf8.csv"}
5xbモードで新規作成しているため、既存のcustomers_utf8.csvは変更されません。
終了コードを後続処理の条件にする
| 終了コード | 意味 | 対応 |
|---|---|---|
| 0 | 正規化成功 | 出力CSVを次工程へ渡す |
| 2 | 入力を読めない、または引数エラー | パス、権限、引数を確認する |
| 3 | 文字コードを受け入れられない | 送信元の出力設定とBOMを確認する |
| 4 | CSV構文または列数が不正 | ファイルを隔離して内容を確認する |
| 5 | 出力できない | 既存ファイル、親ディレクトリ、権限を確認する |
入力ファイルの読み込みに失敗した場合は、終了コード2と1行のJSONを返します。一方、引数不足や未知のオプションをargparseが処理開始前に検出した場合は、同じ終了コード2でもJSONではなく使用法が標準エラー出力へ表示されます。--helpは使用法を表示し、終了コード0で終了します。
この実装で意図的に行わないこと
- EUC-JPなど、契約にない文字コードの推測
- UTF-16またはUTF-32からの自動変換
- 複数のUTF-8 BOMを除去して処理を継続すること
- 改行コード、区切り文字、引用符の正規化
- 日付、金額、必須項目など業務スキーマの検証
- 既存ファイルの上書き
- 大容量CSVのストリーミング処理
このコードは入力全体をメモリへ読み込むため、大容量CSVには向きません。まず小さな業務ファイルの入口処理として使い、容量が大きい場合は上限を決めたうえで別の実装を検討します。
CP932としてデコードできることは、入力元が本当にCP932を採用している証明にはなりません。別の文字コードのバイト列でも偶然デコードできる可能性があるため、--allow-cp932は入力元単位で設定します。ASCII文字だけの入力はUTF-8と同じバイト表現なので、UTF-8として扱われます。
書き込み中に通常のエラーが発生した場合、コードは作成途中のファイルの削除を試みます。ただし、削除自体に失敗する場合や、プロセスの強制終了、OS障害まで含む原子的な保存は保証しません。終了コード0を確認したファイルだけを次工程へ渡します。
実務へ組み込むときの確認項目
- 入力元ごとに、許可する文字コードを記録する
- 下流処理には原本ではなく正規化後のパスを渡す
- 終了コード0のときだけスキーマ移行や集計を開始する
- 標準出力と標準エラー出力を分けて保存する
- 原本を保管する場合は、保存期間とアクセス権を業務ルールで決める
文字コードの問題を入口で分離すれば、後段のスキーマ移行や品質検査はUTF-8だけを前提にできます。対応形式をむやみに増やさず、受け入れ条件を狭く明示することが、再現可能なCSV処理につながります。