nik/newdwh2021
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix:投入データのデータ項目順序が変更されても正常にデータ登録ができるように修正した

Browse Source
This commit is contained in:
*lcOeIaePm0 2021-11-12 10:08:00 +09:00
parent ee365ff0de
commit 938a095330
1 changed files with 45 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

62
ecs/Dockerfile/dataimport/main.py
View File

@ -9,9 +9,6 @@ from common import debug_log
# 定数
DIRECTORY_WORK = '/work/'
DELETE_FLG = 'delete_flg'
UPDATE_USER = 'upd_user'
UPDATE_DATE = 'upd_date'
LOG_LEVEL = {"i": 'Info', "e": 'Error', "w": 'Warning'}
SETTINGS_ITEM = {
'dataSource': 0,
@ -114,13 +111,23 @@ def main(bucket_name, target_data_source, target_file_name, settings_key, db_inf
process_count += 1
# SQL文生成
sql = f'INSERT INTO {settings_list[SETTINGS_ITEM["loadSchemaName"]]} VALUES ('
settings_db_columu_list = settings_list[SETTINGS_ITEM["dbColumuName"]].rstrip().split(',')
sql = f'INSERT INTO {settings_list[SETTINGS_ITEM["loadSchemaName"]]} ('
for i in range(len(settings_db_columu_list)):
sql = f'{sql} {settings_db_columu_list[i]},'
sql = f'{sql} file_name,' # システム項目：取込ファイル名
sql = f'{sql} file_row_cnt,' # システム項目：取込ファイル行番号
sql = f'{sql} delete_flg,' # システム項目：論理削除フラグ
sql = f'{sql} ins_user,' # システム項目：登録者
sql = f'{sql} ins_date,' # システム項目：登録日時
sql = f'{sql} upd_user,' # システム項目：更新者
sql = f'{sql} upd_date)' # システム項目：更新日時
sql = f'{sql} VALUES ('
for i in range(len(line)):
replace_line = line[i].replace('\\', '\\\\')
sql = f'{sql} "{replace_line}",'
sql = f'{sql} "{target_file_name}",' # システム項目：取込ファイル名
sql = f'{sql} "{index}",' # システム項目：取込ファイル行番号
sql = f'{sql} "{index + 1}",' # システム項目：取込ファイル行番号
sql = f'{sql} "0",' # システム項目：論理削除フラグ
sql = f'{sql} CURRENT_USER(),' # システム項目：登録者
sql = f'{sql} CURRENT_TIMESTAMP(),' # システム項目：登録日時
@ -129,6 +136,8 @@ def main(bucket_name, target_data_source, target_file_name, settings_key, db_inf
index += 1
debug_log(sql, log_info, mode)
# ロードスキーマのトランザクション開始
with conn.cursor() as cur:
cur.execute(sql)
@ -148,20 +157,39 @@ def main(bucket_name, target_data_source, target_file_name, settings_key, db_inf
print(f'{datetime.now():%Y-%m-%d %H:%M:%S} {log_info} {LOG_LEVEL["i"]} I-MAIN-08 - ロードスキーマ（{settings_list[SETTINGS_ITEM["loadSchemaName"]]}）のデータを蓄積スキーマ（{settings_list[SETTINGS_ITEM["storageSchemaName"]]}）に登録します')
# SQL文生成
sql = f'INSERT INTO {settings_list[SETTINGS_ITEM["storageSchemaName"]]}'
sql = f'{sql} SELECT t.*'
sql = f'INSERT INTO {settings_list[SETTINGS_ITEM["storageSchemaName"]]} ('
for i in range(len(settings_db_columu_list)):
sql = f'{sql} {settings_db_columu_list[i]},'
sql = f'{sql} file_name,' # システム項目：取込ファイル名
sql = f'{sql} file_row_cnt,' # システム項目：取込ファイル行番号
sql = f'{sql} delete_flg,' # システム項目：論理削除フラグ
sql = f'{sql} ins_user,' # システム項目：登録者
sql = f'{sql} ins_date,' # システム項目：登録日時
sql = f'{sql} upd_user,' # システム項目：更新者
sql = f'{sql} upd_date)' # システム項目：更新日時
sql = f'{sql} SELECT'
for i in range(len(settings_db_columu_list)):
sql = f'{sql} t.{settings_db_columu_list[i]},'
sql = f'{sql} t.file_name,' # システム項目：取込ファイル名
sql = f'{sql} t.file_row_cnt,' # システム項目：取込ファイル行番号
sql = f'{sql} t.delete_flg,' # システム項目：論理削除フラグ
sql = f'{sql} t.ins_user,' # システム項目：登録者
sql = f'{sql} t.ins_date,' # システム項目：登録日時
sql = f'{sql} t.upd_user,' # システム項目：更新者
sql = f'{sql} t.upd_date' # システム項目：更新日時
sql = f'{sql} FROM {settings_list[SETTINGS_ITEM["loadSchemaName"]]} as t'
sql = f'{sql} ON DUPLICATE KEY UPDATE'
settings_db_columu_list = settings_list[SETTINGS_ITEM["dbColumuName"]].rstrip().split(',')
for i in range(len(settings_db_columu_list)):
if settings_db_columu_list[i] == DELETE_FLG:
sql = f'{sql} {settings_db_columu_list[i]}={settings_list[SETTINGS_ITEM["storageSchemaName"]]}.{settings_db_columu_list[i]},' # システム項目：論理削除フラグ
elif settings_db_columu_list[i] == UPDATE_USER:
sql = f'{sql} {settings_db_columu_list[i]}={settings_list[SETTINGS_ITEM["storageSchemaName"]]}.{settings_db_columu_list[i]},' # システム項目：更新者
elif settings_db_columu_list[i] == UPDATE_DATE:
sql = f'{sql} {settings_db_columu_list[i]}={settings_list[SETTINGS_ITEM["storageSchemaName"]]}.{settings_db_columu_list[i]}' # システム項目：更新日時
else:
sql = f'{sql} {settings_db_columu_list[i]}=t.{settings_db_columu_list[i]},'
sql = f'{sql} {settings_db_columu_list[i]}=t.{settings_db_columu_list[i]},'
sql = f'{sql} file_name=t.file_name,' # システム項目：取込ファイル名
sql = f'{sql} file_row_cnt=t.file_row_cnt,' # システム項目：取込ファイル行番号
sql = f'{sql} delete_flg={settings_list[SETTINGS_ITEM["storageSchemaName"]]}.delete_flg,' # システム項目：論理削除フラグ
sql = f'{sql} ins_user=t.ins_user,' # システム項目：登録者
sql = f'{sql} ins_date=t.ins_date,' # システム項目：登録日時
sql = f'{sql} upd_user={settings_list[SETTINGS_ITEM["storageSchemaName"]]}.upd_user,' # システム項目：更新者
sql = f'{sql} upd_date={settings_list[SETTINGS_ITEM["storageSchemaName"]]}.upd_date' # システム項目：更新日時
debug_log(sql, log_info, mode)
# トランザクション開始
print(f'{datetime.now():%Y-%m-%d %H:%M:%S} {log_info} {LOG_LEVEL["i"]} I-MAIN-09 - 標準SQL：{settings_list[SETTINGS_ITEM["storageSchemaName"]]} のトランザクションを開始します')