CSVのデータをPythonでValidation

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

## 要件を整理
チェックしたい内容

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

その他

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

## スクリプトの構成
1. 入力チェックエラーのクラス
2. ログ出力のためロガー
3. 設定ファイル読み込み
↑ここまで準備処理
4. CSVファイル読み込み
5. 行ごとにループしないチェック
6. 行ごとにループするチェック
7. チェックNGはログ出力

## スクリプト作成
###入力チェックエラーのクラス
他のExceptionと区別して処理できるように、CSVチェックのエラーはこのクラスを使う。

```py
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にチェック結果を出力するので、１回実行する度にログを分けた方が見やすいと思い、ログファイル名に現在日時を。

```py
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. 設定ファイル読み込み
設定ファイル

```ini: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に

```py
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 を使います。

```py
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を作成しておく。

```py
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()　で列方向にループ。

```py
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 で囲む。

```py
    # チェック対象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を作成しておく。

```py

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行目が最大桁数。今回は最小と最大に同じ値を設定しているが、異なる値を設定しても良い。

```py
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を作成しておく。
列名２ のデータに"@"が含まれてたらエラーになる。
今回は必要なかったけど、正規表現対応した方がより良い。

```py
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” が含まれない場合エラー。

```py
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 以外が設定されていたらエラー。

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

| 列名１ | 列名２ | 列名３ | 列名４ |
|:-------|:------|:-------|:-------|
| email |  |  |  |
format.csv は Excel で見たら上のようになるCSVを作成しておく。
列名１ のデータに メールアドレス 以外が設定されていたらエラー。
メールアドレスの形式チェックは validate_email というpkgを使用。

```text
pip install validate_email
```

で validate_email をインストールしておく。

```py
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。](https://github.com/ping2shi2/Py2_CSVValidation/blob/master/src/csvValidator.py)