17
26

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

More than 5 years have passed since last update.

CSVのデータをPythonでValidation

Posted at

CSVファイルのデータチェックする用事があったので、Pythonでやった忘備録。

こんなもの必要か?

大抵のサービスやアプリには、サービス開始するために、CSVなんかでデータを一括インポートしてデータ移行する機能がついてます。当然、CSVで一括インポートする機能にはVlidationも実装されてます。
しかし、大量にデータを登録して大量にエラーがでたら、新しいサービスを使い始める計画が狂ってしまう。そこで、CSVのデータを事前にチェックするスクリプトを作ることに。
本当はデータを一括登録する機能には、チェックだけしてデータ登録しない機能が必要なのかも…。
Excelで関数を駆使すればできる内容だけど、Pythonの勉強もかねて。

要件を整理

チェックしたい内容

  • CSVのヘッダに必要な列があるか?
  • ユニークチェック(データの主キーに該当する列に重複がないか?)
  • 必須チェック
  • 桁数チェック
  • 禁止文字チェック(;区切りで複数の禁止文字対応)
  • 指定した文字列を含むこと(禁止文字の反対)
  • 選択値チェック(候補値以外の値が入ってないか?)
  • 形式チェック(今回はメールアドレスか?)

その他

  • 使いまわせるようにチェック仕様をファイルで読み込む
  • 各チェックをする/しないを指定できるように
    (今回は必須チェックと桁数チェックして他のチェックはしない。みたいな…)

スクリプトの構成

  1. 入力チェックエラーのクラス
  2. ログ出力のためロガー
  3. 設定ファイル読み込み
    ↑ここまで準備処理
  4. CSVファイル読み込み
  5. 行ごとにループしないチェック
  6. 行ごとにループするチェック
  7. チェック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にチェック結果を出力するので、1回実行する度にログを分けた方が見やすいと思い、ログファイル名に現在日時を。

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 のイメージ

列名1 列名2 列名3 列名4
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 列名2 列名3 列名4
1
"1" が設定されてる 列名2 のデータを重複チェックする。
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 列名2 列名3 列名4
1 1 1
"1" が設定されてる 列名1、列名2、列名4 のデータを必須チェックする。
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 のイメージ。

列名1 列名2 列名3 列名4
5
5
length.csv は Excel で見たら上のようになるCSVを作成しておく。
データ部の1行目に"5"が設定されてる 列名2 が5桁以上か?のチェックで、データ部の2行目に"5"が設定されている 列名2 が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 のイメージ。

列名1 列名2 列名3 列名4
@
illegal.csv は Excel で見たら上のようになるCSVを作成しておく。
列名2 のデータに"@"が含まれてたらエラーになる。
今回は必要なかったけど、正規表現対応した方がより良い。
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 のイメージ。

列名1 列名2 列名3 列名4
google.com
include.csv は Excel で見たら上のようになるCSVを作成しておく。
列名1 のデータに文字列 ”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 のイメージ。

列名1 列名2 列名3 列名4
True
False
select.csv は Excel で見たら上のようになるCSVを作成しておく。
列名4 のデータに 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 のイメージ。

列名1 列名2 列名3 列名4
email
format.csv は Excel で見たら上のようになるCSVを作成しておく。
列名1 のデータに メールアドレス 以外が設定されていたらエラー。
メールアドレスの形式チェックは 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。

17
26
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
17
26

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?