105 lines
4.8 KiB
Python
105 lines
4.8 KiB
Python
from datetime import datetime
|
||
import boto3
|
||
import io
|
||
import csv
|
||
from error import error
|
||
from common import debug_log
|
||
|
||
# 定数
|
||
DIRECTORY_WORK = '/work/'
|
||
LOG_LEVEL = {'i': 'Info', 'e': 'Error'}
|
||
SETTINGS_ITEM = {
|
||
'dataSource': 0,
|
||
'delimiter': 1,
|
||
'charCode': 2,
|
||
'quotechar': 3,
|
||
'lineFeedCode': 4,
|
||
'headerFlag': 5,
|
||
'csvNumItems': 6,
|
||
'csvNameItems': 7,
|
||
'dbColumuName': 8,
|
||
'storageSchemaName': 9,
|
||
'loadSchemaName': 10,
|
||
'exSqlFileName': 11,
|
||
}
|
||
LINE_FEED_CODE = {
|
||
'CR': '\r',
|
||
'LF': '\n',
|
||
'CRLF': '\r\n',
|
||
}
|
||
|
||
# クラス変数
|
||
s3_resource = boto3.resource('s3')
|
||
|
||
|
||
# チェック例外クラス
|
||
class CheckError(Exception):
|
||
pass
|
||
|
||
|
||
def check(bucket_name, target_data_source, target_file_name, settings_key, log_info, mode):
|
||
"""チェック処理
|
||
Args:
|
||
bucket_name : バケット名
|
||
target_data_source : 投入データのディレクトリ名よりデータソースに該当する部分
|
||
target_file_name : 投入データのファイル名
|
||
settings_key : 投入データに該当する個別設定ファイルのフルパス
|
||
log_info : ログに記載するデータソース名とファイル名
|
||
mode : 処理モード
|
||
Raises:
|
||
CheckError : チェックでエラーがあった場合に発生する例外
|
||
"""
|
||
|
||
try:
|
||
debug_log(f'引数 bucket_name : {bucket_name}', log_info, mode)
|
||
debug_log(f'引数 target_data_source : {target_data_source}', log_info, mode)
|
||
debug_log(f'引数 target_file_name : {target_file_name}', log_info, mode)
|
||
debug_log(f'引数 settings_key : {settings_key}', log_info, mode)
|
||
debug_log(f'引数 log_info : {log_info}', log_info, mode)
|
||
debug_log(f'引数 mode : {mode}', log_info, mode)
|
||
|
||
# ① チェック処理開始ログを出力する
|
||
print(f'{datetime.now():%Y-%m-%d %H:%M:%S} {log_info} {LOG_LEVEL["i"]} I-CHK-01 - チェック処理を開始します')
|
||
|
||
# データ読込
|
||
settings_obj = s3_resource.Object(bucket_name, settings_key)
|
||
settings_response = settings_obj.get()
|
||
settings_list = []
|
||
for line in io.TextIOWrapper(io.BytesIO(settings_response["Body"].read()), encoding='utf-8'):
|
||
settings_list.append(line.rstrip('\n'))
|
||
|
||
work_key = target_data_source + DIRECTORY_WORK + target_file_name
|
||
work_obj = s3_resource.Object(bucket_name, work_key)
|
||
work_response = work_obj.get()
|
||
work_data = io.TextIOWrapper(io.BytesIO(work_response["Body"].read()), encoding=settings_list[SETTINGS_ITEM["charCode"]], newline=LINE_FEED_CODE[settings_list[SETTINGS_ITEM["lineFeedCode"]]])
|
||
for line in csv.reader(work_data, quotechar=settings_list[SETTINGS_ITEM["quotechar"]], delimiter=settings_list[SETTINGS_ITEM["delimiter"]]):
|
||
work_header_list = line
|
||
break
|
||
|
||
# ② C-1の項目数チェックを開始する
|
||
print(f'{datetime.now():%Y-%m-%d %H:%M:%S} {log_info} {LOG_LEVEL["i"]} I-CHK-02 - C-1のチェックを開始します')
|
||
work_header_list_len = len(work_header_list)
|
||
if work_header_list_len == int(settings_list[SETTINGS_ITEM["csvNumItems"]]):
|
||
print(f'{datetime.now():%Y-%m-%d %H:%M:%S} {log_info} {LOG_LEVEL["i"]} I-CHK-03 - C-1：正常終了')
|
||
else:
|
||
raise CheckError(f'E-CHK-01 - 項目数が一致しません 個別設定ファイル項目数：{settings_list[SETTINGS_ITEM["csvNumItems"]]} 投入データ項目数：{work_header_list_len}')
|
||
|
||
# ③ C-2の項目並び順チェック開始する
|
||
if int(settings_list[SETTINGS_ITEM["headerFlag"]]) == True:
|
||
print(f'{datetime.now():%Y-%m-%d %H:%M:%S} {log_info} {LOG_LEVEL["i"]} I-CHK-04 - C-2のチェックを開始します')
|
||
settings_header_list = settings_list[SETTINGS_ITEM["csvNameItems"]].rstrip().split(',')
|
||
for i in range(len(settings_header_list)):
|
||
if not settings_header_list[i] == work_header_list[i]:
|
||
raise CheckError(f'E-CHK-02 - 項目順序が一致しません {i}番目の項目 個別設定ファイル項目：{settings_header_list[i]} 投入データ項目：{work_header_list[i]}')
|
||
print(f'{datetime.now():%Y-%m-%d %H:%M:%S} {log_info} {LOG_LEVEL["i"]} I-CHK-05 - C-2：正常終了')
|
||
|
||
# ④ チェック処理終了ログを出力する
|
||
print(f'{datetime.now():%Y-%m-%d %H:%M:%S} {log_info} {LOG_LEVEL["i"]} I-CHK-06 - チェック処理を終了します')
|
||
except CheckError as e:
|
||
print(f'{datetime.now():%Y-%m-%d %H:%M:%S} {log_info} {LOG_LEVEL["e"]} {e}')
|
||
error(bucket_name, target_data_source, target_file_name, log_info)
|
||
|
||
except Exception as e:
|
||
print(f'{datetime.now():%Y-%m-%d %H:%M:%S} {log_info} {LOG_LEVEL["e"]} E-CHK-99 - エラー内容：{e}')
|
||
error(bucket_name, target_data_source, target_file_name, log_info)
|