fix: 0バイト判定を修正。ヘッダ行は別のところでも使っていたため、判定処理のみ関数化

(本当は他のも関数化したいけど我慢)
This commit is contained in:
shimoda.m@nds-tyo.co.jp 2022-09-16 12:47:14 +09:00
parent 3a177ae18c
commit cda1a79e7a

View File

@ -1,11 +1,13 @@
from datetime import datetime
import boto3
import io
import csv
import io
import sys
from datetime import datetime
import boto3
from common import convert_quotechar, debug_log
from end import end
from error import error
from common import debug_log, convert_quotechar
# 定数
DIRECTORY_WORK = '/work/'
@ -83,17 +85,13 @@ def check(bucket_name, target_data_source, target_file_name, settings_key, log_i
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"]]])
work_header_list = []
# ヘッダ行のみのファイルは、0バイトファイル同等とみなす
for i, line in enumerate(csv.reader(work_data, quotechar=convert_quotechar(settings_list[SETTINGS_ITEM["quotechar"]]), delimiter=settings_list[SETTINGS_ITEM["delimiter"]])):
if int(settings_list[SETTINGS_ITEM["headerFlag"]]) == True and i == 0:
work_header_list = line
continue
work_header_list = line
break
# ② C-0のデータ件数チェックを開始する
print(f'{datetime.now():%Y-%m-%d %H:%M:%S} {log_info} {LOG_LEVEL["i"]} I-CHK-02 - C-0のチェックを開始します')
if not len(work_header_list):
if is_empty_file(work_header_list, settings_list):
print(f'{datetime.now():%Y-%m-%d %H:%M:%S} {log_info} {LOG_LEVEL["i"]} I-CHK-03 - 投入ファイルが0バイトのため処理を終了します')
end(bucket_name, target_data_source, target_file_name, '', log_info, mode)
print(f'{datetime.now():%Y-%m-%d %H:%M:%S} {log_info} {LOG_LEVEL["i"]} I-CHK-04 - 終了処理完了')
@ -126,3 +124,22 @@ def check(bucket_name, target_data_source, target_file_name, settings_key, log_i
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)
def is_empty_file(work_header_list: list, settings_list: list):
"""② C-0のデータ件数チェック
ヘッダ行がある場合は1行目を読み飛ばして判定する
Args:
work_header_list (list): CSVファイルの1行目
settings_list (list): 個別設定ファイルのリスト
Returns:
bool: CSVファイルの1行目が0件だった場合はTrue
"""
has_header = settings_list[SETTINGS_ITEM["headerFlag"]]
# ヘッダのみのファイルも0バイトファイルをみなす
if has_header:
return len(work_header_list[1:]) == 0
return len(work_header_list) == 0