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

feat: ログ追加.変数名見直し

Browse Source
This commit is contained in:
shimoda.m@nds-tyo.co.jp 2025-05-07 18:04:33 +09:00
parent f471a2d444
commit e31b20692c
1 changed files with 22 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
ecs/dataimport/dataimport/chk.py
View File

@ -94,15 +94,16 @@ def check(bucket_name, target_data_source, target_file_name, settings_key, log_i
work_key = target_data_source + DIRECTORY_WORK + target_file_name
work_obj_response = s3_client.get_object(
Bucket=bucket_name, Key=work_key)
work_obj_bytes = work_obj_response["Body"].read()
work_data_file = work_obj_bytes
work_obj_file = work_obj_response["Body"].read()
work_data_bytes = work_obj_file
compressed_flag = settings_list[SETTINGS_ITEM["compressedFlag"]]
# 圧縮されている場合は解凍する
if compressed_flag and compressed_flag == '1':
work_data_file = uncompress_file(work_data_file, settings_list)
work_data_bytes = uncompress_file(
work_data_bytes, settings_list, log_info)
work_data = io.TextIOWrapper(io.BytesIO(work_data_file), encoding=settings_list[SETTINGS_ITEM["charCode"]],
work_data = io.TextIOWrapper(io.BytesIO(work_data_bytes), encoding=settings_list[SETTINGS_ITEM["charCode"]],
newline=LINE_FEED_CODE[settings_list[SETTINGS_ITEM["lineFeedCode"]]])
work_csv_row = []
for i, line in enumerate(csv.reader(work_data, quotechar=convert_quotechar(settings_list[SETTINGS_ITEM["quotechar"]]),
@ -186,8 +187,21 @@ def is_empty_file(work_csv_row: list, settings_list: list):
return len(work_csv_row) == 0
def uncompress_file(work_data_file: bytes, settings_list: list):
if settings_list[SETTINGS_ITEM['compression']] == 'zip':
def uncompress_file(work_data_file: bytes, settings_list: list, log_info) -> bytes:
"""指定された形式で圧縮ファイルを展開し、ローカルに書き出す。
Args:
work_data_file (bytes): S3から読み込んだ登録
settings_list (list): 個別設定ファイルのリスト
log_info (str): ログに記載するデータソース名とファイル名
Returns:
bytes: 解凍後のファイルのバイト配列
"""
compression = settings_list[SETTINGS_ITEM['compression']]
print(
f'{datetime.now():%Y-%m-%d %H:%M:%S} {log_info} {LOG_LEVEL["i"]} I-CHK-11 - 投入ファイルが圧縮されているため、展開処理を行います。圧縮形式：{compression}')
if compression == 'zip':
file_bytes = None
with zipfile.ZipFile(io.BytesIO(work_data_file), 'r') as zip_ref:
for file_name in zip_ref.namelist():
@ -206,6 +220,8 @@ def uncompress_file(work_data_file: bytes, settings_list: list):
delimiter=settings_list[SETTINGS_ITEM["delimiter"]])
for row in csv_reader:
csv_writer.writerow(row)
print(
f'{datetime.now():%Y-%m-%d %H:%M:%S} {log_info} {LOG_LEVEL["i"]} I-CHK-12 - 投入ファイルの展開が正常終了しました')
return file_bytes
# TODO: zip以外の圧縮形式に対応する際に追記すること
else: