CSVファイルのデータチェックする用事があったので、Pythonでやった忘備録。
こんなもの必要か？
大抵のサービスやアプリには、サービス開始するために、CSVなんかでデータを一括インポートしてデータ移行する機能がついてます。当然、CSVで一括インポートする機能にはVlidationも実装されてます。
しかし、大量にデータを登録して大量にエラーがでたら、新しいサービスを使い始める計画が狂ってしまう。そこで、CSVのデータを事前にチェックするスクリプトを作ることに。
本当はデータを一括登録する機能には、チェックだけしてデータ登録しない機能が必要なのかも…。
Excelで関数を駆使すればできる内容だけど、Pythonの勉強もかねて。
要件を整理
チェックしたい内容
- CSVのヘッダに必要な列があるか？
- ユニークチェック（データの主キーに該当する列に重複がないか？）
- 必須チェック
- 桁数チェック
- 禁止文字チェック（;区切りで複数の禁止文字対応）
- 指定した文字列を含むこと（禁止文字の反対）
- 選択値チェック（候補値以外の値が入ってないか？）
- 形式チェック（今回はメールアドレスか？）
その他
- 使いまわせるようにチェック仕様をファイルで読み込む
- 各チェックをする／しないを指定できるように （今回は必須チェックと桁数チェックして他のチェックはしない。みたいな…）
スクリプトの構成
- 入力チェックエラーのクラス
- ログ出力のためロガー
- 設定ファイル読み込み ↑ここまで準備処理
- CSVファイル読み込み
- 行ごとにループしないチェック
- 行ごとにループするチェック
- チェックNGはログ出力
スクリプト作成
入力チェックエラーのクラス
他のExceptionと区別して処理できるように、CSVチェックのエラーはこのクラスを使う。
class CsvValidationError (Exception):
"""
csvファイルのValidation時に発生するエラー
"""
def __init__ (self, message):
if isinstance(message, unicode):
super(CsvValidationError, self).__init__(message.encode('utf-8'))
self.message = message
elif isinstance(message, str):
super(CsvValidationError, self).__init__(message)
self.message = message.decode('utf-8')
else:
raise TypeError
def __unicode__ (self):
# エラーメッセージ
return (u'CSVチェックエラー "{0}"'.format (self.message))
日本語を扱おうとしてソースが見にくくなってる気が…、そろそろPython3にした方が良いかな。
2. ログ出力のためロガー
logにチェック結果を出力するので、１回実行する度にログを分けた方が見やすいと思い、ログファイル名に現在日時を。
import datetime
def setup_logger():
"""
Setup Logger.
"""
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
log_file_path = os.path.join(os.path.abspath('..'), 'log')
d = datetime.datetime.now()
log_file_path = os.path.join(log_file_path, 'CSVValidator' + d.strftime("%Y%m%d %H%M%S") + '.log')
fh = logging.FileHandler(log_file_path, encoding = "UTF-8")
logger.addHandler(fh)
sh = logging.StreamHandler()
logger.addHandler(sh)
formatter = logging.Formatter(
'%(asctime)s:%(lineno)d:%(levelname)s:%(message)s')
fh.setFormatter(formatter)
sh.setFormatter(formatter)
return logger
logger = setup_logger()
3. 設定ファイル読み込み
設定ファイル
settings.ini
[settings]
#ヘッダチェック
VALIDATE_HEADER : False
HEADER_FILE_NAME : header.csv
#必須チェック
VALIDATE_REQUIRE : True
REQUIRE_FILE_NAME : require.csv
#桁数チェック
VALIDATE_LENGTH : True
LENGTH_FILE_NAME : length.csv
#禁止文字チェックファイル
VALIDATE_ILLEGAL_CHAR : True
ILLEGAL_CHAR_FILE_NAME : illegal_char.csv
#形式チェックファイル
VALIDATE_FORMAT : True
FORMAT_FILE_NAME : format.csv
#選択チェックファイル
VALIDATE_SELECT : False
SELECT_FILE_NAME : select.csv
#ユニークチェック
VALIDATE_UNIQUE : True
UNIQUE_FILE_NAME : unique.csv
#文字列含むチェック
VALIDATE_INCLUDE : True
INCLUDE_FILE_NAME : include.csv
#Validation対象ファイル
OBJECT_FILE_NAME : 対象データ.csv
この設定ファイルを読みこんで、どこからでも使えるようにglobalに
def setup_config():
"""
Setup Config.
"""
# 設定ファイル読み込み
conf_dir_path = os.path.join(os.path.abspath('..'), 'conf')
conf_file_path = os.path.join(conf_dir_path, 'settings.ini')
conf_file = ConfigParser.SafeConfigParser()
conf_file.read(conf_file_path)
# configの値はglobal変数に
global VALIDATE_HEADER
global VALIDATE_REQUIRE
global VALIDATE_LENGTH
global VALIDATE_ILLEGAL_CHAR
global VALIDATE_FORMAT
global VALIDATE_SELECT
global VALIDATE_UNIQUE
global VALIDATE_INCLUDE
global HEADER_FILE_NAME
global REQUIRE_FILE_NAME
global LENGTH_FILE_NAME
global ILLEGAL_CHAR_FILE_NAME
global FORMAT_FILE_NAME
global SELECT_FILE_NAME
global UNIQUE_FILE_NAME
global INCLUDE_FILE_NAME
global OBJECT_FILE_NAME
# 設定ファイルからチェック種類を取得
VALIDATE_HEADER = conf_file.getboolean(
'settings', 'VALIDATE_HEADER')
VALIDATE_REQUIRE = conf_file.getboolean(
'settings', 'VALIDATE_REQUIRE')
VALIDATE_LENGTH = conf_file.getboolean(
'settings', 'VALIDATE_LENGTH')
VALIDATE_ILLEGAL_CHAR = conf_file.getboolean(
'settings', 'VALIDATE_ILLEGAL_CHAR')
VALIDATE_FORMAT = conf_file.getboolean(
'settings', 'VALIDATE_FORMAT')
VALIDATE_SELECT = conf_file.getboolean(
'settings', 'VALIDATE_SELECT')
VALIDATE_UNIQUE = conf_file.getboolean(
'settings', 'VALIDATE_UNIQUE')
VALIDATE_INCLUDE = conf_file.getboolean(
'settings', 'VALIDATE_INCLUDE')
# 設定ファイルから各ファイル名を取得
if VALIDATE_HEADER:
HEADER_FILE_NAME = unicode(conf_file.get(
'settings', 'HEADER_FILE_NAME'), 'UTF-8')
if VALIDATE_REQUIRE:
REQUIRE_FILE_NAME = unicode(conf_file.get(
'settings', 'REQUIRE_FILE_NAME'), 'UTF-8')
if VALIDATE_LENGTH:
LENGTH_FILE_NAME = unicode(conf_file.get(
'settings', 'LENGTH_FILE_NAME'), 'UTF-8')
if VALIDATE_ILLEGAL_CHAR:
ILLEGAL_CHAR_FILE_NAME = unicode(conf_file.get(
'settings', 'ILLEGAL_CHAR_FILE_NAME'), 'UTF-8')
if VALIDATE_FORMAT:
FORMAT_FILE_NAME = unicode(conf_file.get(
'settings', 'FORMAT_FILE_NAME'), 'UTF-8')
if VALIDATE_SELECT:
SELECT_FILE_NAME = unicode(conf_file.get(
'settings', 'SELECT_FILE_NAME'), 'UTF-8')
if VALIDATE_UNIQUE:
UNIQUE_FILE_NAME = unicode(conf_file.get(
'settings', 'UNIQUE_FILE_NAME'), 'UTF-8')
if VALIDATE_INCLUDE:
INCLUDE_FILE_NAME = unicode(conf_file.get(
'settings', 'INCLUDE_FILE_NAME'), 'UTF-8')
OBJECT_FILE_NAME = unicode(conf_file.get(
'settings', 'OBJECT_FILE_NAME'), 'UTF-8')
setup_config()
んー、長いしカッコ悪い。そのうちカッコ良く書けるようにしよう。
4. CSVファイル読み込み
最近お気に入りの pandas を使います。
import pandas as pd
# CSVファイルディレクトリ
csv_files_path = os.path.join(os.path.abspath('..'), 'csvfiles')
# ヘッダの定義
if VALIDATE_HEADER:
header_csv_file = os.path.join(csv_files_path, HEADER_FILE_NAME)
header_df = pd.read_csv(header_csv_file, encoding = 'utf8')
# 必須項目の定義
if VALIDATE_REQUIRE:
require_csv_file = os.path.join(csv_files_path, REQUIRE_FILE_NAME)
require_df = pd.read_csv(require_csv_file, encoding = 'utf8')
# settings.iniで定義したファイルをひたすら読み込む。長いので途中省略
# Vlidation対象CSV読み込み
object_csv_file = os.path.join(csv_files_path, OBJECT_FILE_NAME)
object_df = pd.read_csv(object_csv_file, encoding = 'utf8')
5. 行ごとにループしないチェック
・CSVのヘッダに必要な列があるか？
header_df のイメージ
| 列名１ | 列名２ | 列名３ | 列名４ |
|---|
header.csv は Excel で見たら上のようになるCSVを作成しておく。
def validate_header(header_df, object_df):
"""
validate header
"""
# 1.ヘッダの定義をlistに変換
correct_header_list = header_df.columns.values.tolist()
# 2.チェック対象のCSVヘッダをlistに変換
object_header_list = object_df.columns.values.tolist()
# set(1) - set(2) てやるとlistの差分が？！
shortage_header_set = set(correct_header_list) - set(object_header_list)
error_list = list(shortage_header_set)
if error_list:
# 差分があれば自作エラークラスにほうりこんで raise
raise CsvValidationError(u"CSVヘッダエラー : " + ','.join(error_list))
・ ユニークチェック
主キーになる列に重複がないか？のチェック
重複チェックをする列名を定義した unique_df のイメージ。
| 列名１ | 列名２ | 列名３ | 列名４ |
|---|---|---|---|
| 1 |
"1" が設定されてる 列名２ のデータを重複チェックする。
unique.csv は Excel で見たら上のようになるCSVを作成しておく。
df.iteritems() で列方向にループ。
def validate_unique(unique_df, object_df):
"""
validate unique
"""
error_list = []
# 重複チェック定義のDataFrameを列方向にループ
for column, value in unique_df.iteritems():
if pd.isnull(value[0]):
# 値が設定されてなければ次の列へ
continue
# 該当列を重複判定
dup_series = object_df[column].duplicated()
# 重複結果Trueのindexを取得
dup_ix_list = (dup_series[dup_series == True].index)
# +2 してcsvの行番号に変換
dup_ix_list = map(lambda x: x+2, dup_ix_list)
error_list.append(u"列=" + column + " : Line No. = " + ','.join(map(str,dup_ix_list)))
if dup_ix_list:
raise CsvValidationError(u"重複エラー : " + ','.join(error_list))
6. 行ごとにループするチェック
df.iterrows()で行方向にループする。series_row はdfの1行分のデータが入った変数。
ループの中で、settings.ini で「チェックする。」としたチェック処理を実行する。
チェック処理は、try except で囲む。
# チェック対象CSVを行方向にループ
for i, series_row in object_df.iterrows():
# iをCSVの行数に変換
line_no = i + 2
if VALIDATE_REQUIRE:
try:
validate_require(require_df, series_row)
except CsvValidationError as e:
# 自作のCSVチェックエラーは、ログにエラー内容を出力
logger.error(u"Line : " + unicode(line_no) + " : " + e.message)
except Exception as e:
# CSVチェックエラー以外は、エラーが発生したCSVの行とtracebackを出力
logger.error(u"Line :" + unicode(line_no))
logger.exception(e)
・必須チェック
必須チェックをする列名を定義した require_df のイメージ。
| 列名１ | 列名２ | 列名３ | 列名４ |
|---|---|---|---|
| 1 | 1 | 1 |
"1" が設定されてる 列名１、列名２、列名４ のデータを必須チェックする。
require.csv は Excel で見たら上のようになるCSVを作成しておく。
def validate_require(require_df, series_row):
"""
validate require
"""
error_list = []
for column, value in require_df.iteritems():
if value.values[0:1][0] == 1:
# 必須項目の場合
if pd.isnull(series_row[column]):
error_list.append(unicode(column))
if error_list:
raise CsvValidationError(u"必須項目エラー : " + ','.join(error_list))
・桁数チェック
桁数チェックをする列名を定義した length_df のイメージ。
| 列名１ | 列名２ | 列名３ | 列名４ |
|---|---|---|---|
| 5 | |||
| 5 |
length.csv は Excel で見たら上のようになるCSVを作成しておく。
データ部の1行目に"5"が設定されてる 列名２ が5桁以上か？のチェックで、データ部の2行目に"5"が設定されている 列名２ が5桁以下か？をチェックする。1行目が最小桁数で2行目が最大桁数。今回は最小と最大に同じ値を設定しているが、異なる値を設定しても良い。
def validate_length(length_df, series_row):
"""
validate require
"""
error_list = []
for column, value in length_df.iteritems():
min_length = value.values[0:1][0]
if min_length > 0:
# minlengthcheck対象の場合
if not len(str(series_row[column])) >= min_length:
error_list.append(column)
max_length = value.values[0:2][0]
if max_length > 0:
# maxlengthcheck対象の場合
if not len(str(series_row[column])) <= max_length:
error_list.append(column)
if error_list:
raise CsvValidationError(u"項目長エラー : " + ','.join(error_list))
・禁止文字チェック
禁止文字チェックをする列名を定義した illegal_char_df のイメージ。
| 列名１ | 列名２ | 列名３ | 列名４ |
|---|---|---|---|
| @ |
illegal.csv は Excel で見たら上のようになるCSVを作成しておく。
列名２ のデータに"@"が含まれてたらエラーになる。
今回は必要なかったけど、正規表現対応した方がより良い。
def validate_illegal_char(illegal_char_df, series_row):
"""
validate_illegal_char
"""
error_list = []
for column, value in illegal_char_df.iteritems():
illegal_chars = value.values[0:1][0]
if pd.isnull(illegal_chars):
# NaNの場合次の列へ
continue
# ";"区切りの複数禁止文字を分割
illegal_char_list = illegal_chars.split(";")
# 禁止文字のlistでループ
for illegal_char in illegal_char_list:
if illegal_char in str(series_row[column]):
# 禁止文字が含まれてる場合
error_list.append(column)
if error_list:
raise CsvValidationError(u"禁止文字エラー : " + ','.join(error_list))
・指定した文字列を含むこと（禁止文字の反対）のチェック
禁止文字チェックをする列名を定義した include_df のイメージ。
| 列名１ | 列名２ | 列名３ | 列名４ |
|---|---|---|---|
| google.com |
include.csv は Excel で見たら上のようになるCSVを作成しておく。
列名１ のデータに文字列 ”google.com” が含まれない場合エラー。
def validate_include(include_df, series_row):
"""
validate_include_str
"""
error_list = []
for column, value in include_df.iteritems():
include_strs = value.values[0:1][0]
if pd.isnull(include_strs):
# NaNの場合次の列へ
continue
# ";"区切りの複数formatを分割
include_str_list = include_strs.split(";")
if len(include_str_list) > 0:
if pd.isnull(series_row[column]):
continue
# チェック対象CSVに、";"区切りで複数の値を設定できる仕様のため
object_strs = str(series_row[column])
object_str_list = object_strs.split(";")
for include_str in include_str_list:
# format check
for object_str in object_str_list:
if not include_str in object_str:
error_list.append(column)
if error_list:
raise CsvValidationError(u"キーワード文字列なしエラー : " + ','.join(error_list))
・選択値チェック（候補値以外の値が入ってないか？）
選択値チェックをする列名を定義した select_df のイメージ。
| 列名１ | 列名２ | 列名３ | 列名４ |
|---|---|---|---|
| True | |||
| False |
select.csv は Excel で見たら上のようになるCSVを作成しておく。
列名４ のデータに True/False 以外が設定されていたらエラー。
def validate_select(select_df, series_row):
"""
validate_select
"""
error_list = []
for column, value in select_df.iteritems():
if pd.isnull(value.values[0:1]):
# データなしの場合
continue
# 候補値のlistを作成
select_list = map(str, list(value))
object_value = series_row[column]
if not object_value in select_list:
error_list.append(column)
if error_list:
raise CsvValidationError(u"選択値エラー : " + ','.join(error_list))
・形式チェック（今回はメールアドレスか？）
選択値チェックをする列名を定義した format_df のイメージ。
| 列名１ | 列名２ | 列名３ | 列名４ |
|---|---|---|---|
format.csv は Excel で見たら上のようになるCSVを作成しておく。
列名１ のデータに メールアドレス 以外が設定されていたらエラー。
メールアドレスの形式チェックは validate_email というpkgを使用。
pip install validate_email
で validate_email をインストールしておく。
from validate_email import validate_email
def validate_format(format_df, series_row):
"""
validate_illegal_char
"""
error_list = []
for column, value in format_df.iteritems():
formats = value.values[0:1][0]
if pd.isnull(formats):
# NaNの場合次の列へ
continue
# ";"区切りの複数formatを分割
format_list = formats.split(";")
if len(format_list) > 0:
if pd.isnull(series_row[column]):
continue
# チェック対象CSVに、";"区切りで複数の値を設定できる仕様のため
object_strs = str(series_row[column])
object_str_list = object_strs.split(";")
for format in format_list:
# format check
if format == "email":
for object_str in object_str_list:
if not validate_email(object_str):
error_list.append(column)
if error_list:
raise CsvValidationError(u"データフォーマットエラー : " + ','.join(error_list))
CSVのチェックスクリプトできました。
ソースはGitHub。