From f1ce484d1dda6e911fb9080997312e8e67221699 Mon Sep 17 00:00:00 2001 From: "shimoda.m@nds-tyo.co.jp" Date: Tue, 13 Feb 2024 16:24:16 +0900 Subject: [PATCH] =?UTF-8?q?feat:=20=E3=83=81=E3=82=A7=E3=83=83=E3=82=AF?= =?UTF-8?q?=E5=AF=BE=E8=B1=A1=E5=B9=B4=E6=9C=88=E3=81=AE=E3=83=AD=E3=82=B8?= =?UTF-8?q?=E3=83=83=E3=82=AF=E8=BF=BD=E5=8A=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../encise-data-unreceive-check.py | 91 ++++++++++++------- 1 file changed, 59 insertions(+), 32 deletions(-) diff --git a/lambda/encise-data-unreceive-check/encise-data-unreceive-check.py b/lambda/encise-data-unreceive-check/encise-data-unreceive-check.py index e4e26de8..1667f9e2 100644 --- a/lambda/encise-data-unreceive-check/encise-data-unreceive-check.py +++ b/lambda/encise-data-unreceive-check/encise-data-unreceive-check.py @@ -8,6 +8,7 @@ import re from zoneinfo import ZoneInfo import boto3 +from dateutil.relativedelta import relativedelta # 環境変数 CHECK_BUCKET_NAME = os.environ["CHECK_BUCKET_NAME"] @@ -17,7 +18,8 @@ LOG_LEVEL = os.environ["LOG_LEVEL"] MAIL_TEMPLATE_FOLDER_PATH = os.environ["MAIL_TEMPLATE_FOLDER_PATH"] MBJ_NOTICE_TOPIC = os.environ["MBJ_NOTICE_TOPIC"] PROCESSED_MESSAGE_DYNAMODB_TABLE_NAME = os.environ["PROCESSED_MESSAGE_DYNAMODB_TABLE_NAME"] -PROCESSED_MESSAGE_EXPIRES_PERIOD = int(os.environ["PROCESSED_MESSAGE_EXPIRES_PERIOD"]) +PROCESSED_MESSAGE_EXPIRES_PERIOD = int( + os.environ["PROCESSED_MESSAGE_EXPIRES_PERIOD"]) TZ = os.environ["TZ"] # 定数 @@ -91,7 +93,8 @@ def put_success_messages_to_dynamo_db(batch_item_success: list[str]) -> bool: # レコードの有効期限を算出 now = datetime.datetime.now(ZoneInfo(TZ)) - record_expiration_datetime = now + datetime.timedelta(minutes=PROCESSED_MESSAGE_EXPIRES_PERIOD) + record_expiration_datetime = now + \ + datetime.timedelta(minutes=PROCESSED_MESSAGE_EXPIRES_PERIOD) record_expiration_time = record_expiration_datetime.timestamp() for message_id in batch_item_success: @@ -158,11 +161,13 @@ def encise_data_unreceive_check(records: list, execute_month: str) -> tuple[list message_id = record["messageId"] # 2.DynamoDBテーブルからレコードを取得し、処理済みメッセージかどうかを判別する if is_duplicate_message(message_id): - logger.info(f'受信したメッセージは既に処理済みのため、処理をスキップします。メッセージID: {message_id}') + logger.info( + f'受信したメッセージは既に処理済みのため、処理をスキップします。メッセージID: {message_id}') continue except Exception as e: logger.exception(f"E-02-01 メッセージ重複チェック処理に失敗しました　エラー内容：{e}") - batch_item_failures.append(make_failure_item_on_error(message_id)) + batch_item_failures.append( + make_failure_item_on_error(message_id)) continue # SQSパラメータをJSONシリアライズし、Pythonの辞書オブジェクト(イベントパラメータ)を取得する。 @@ -181,49 +186,60 @@ def encise_data_unreceive_check(records: list, execute_month: str) -> tuple[list ) logger.info('I-03-02 受領チェック対象ファイルリストを読み込みました') except Exception as e: - logger.exception(f"E-03-01 受領チェック対象ファイルリストの読み込みに失敗しました　エラー内容：{e}") - batch_item_failures.append(make_failure_item_on_error(message_id)) + logger.exception( + f"E-03-01 受領チェック対象ファイルリストの読み込みに失敗しました　エラー内容：{e}") + batch_item_failures.append( + make_failure_item_on_error(message_id)) continue # ④ 受領チェック処理を行う receive_timing = event_parameter['receive_timing'] logger.info(f'I-04-01 Enciseデータ受領チェック ({receive_timing}) 処理開始') - object_prefix = f'{event_parameter["check_folder_prefix"]}{execute_month}/' + object_prefix = f'{event_parameter["check_folder_prefix"]}/{execute_month}/' # 1.Enciseデータバックアップ保管バケットの処理稼働月に該当するサブフォルダにあるファイル一覧を取得する logger.info( f'I-04-02 オブジェクトリストの取得　取得先：{CHECK_BUCKET_NAME}/{object_prefix}' ) - check_target_file_list_response = s3_client.list_objects_v2(CHECK_BUCKET_NAME, Prefix=object_prefix) - check_target_file_list = [] - for content in check_target_file_list_response[['Contents']]: + receive_file_list_response = s3_client.list_objects_v2( + Bucket=CHECK_BUCKET_NAME, Prefix=object_prefix) + receive_file_list = [] + for content in receive_file_list_response.get('Contents', []): # オブジェクトのキーからファイル名を切り出してリストに追加 obj_key = content['Key'].rsplit('/', INDEX_SPLIT_NUM) - check_target_file_list.append(obj_key[INDEX_LAST]) + receive_file_list.append(obj_key[INDEX_LAST]) # 2.I/Fファイルチェック処理 logger.info(f'I-04-03 Enciseデータ({receive_timing}) ファイルチェック処理開始') - receive_monthly_file_name_body = io.TextIOWrapper(io.BytesIO(check_target_file_list_response["Body"].read()), encoding='utf-8') + check_target_file_name_body = io.TextIOWrapper(io.BytesIO( + check_target_file_list_response["Body"].read()), encoding='utf-8') match_count = 0 row_count = 0 - for tsv_row in csv.reader(receive_monthly_file_name_body, delimiter='\t'): + for tsv_row in csv.reader(check_target_file_name_body, delimiter='\t'): # 「④１．」で取得したリストが「③」で読み込んだファイル内に存在するか確認する is_file_not_exists = True - for file_name in check_target_file_list: - match_result = re.fullmatch(tsv_row[INDEX_REGEX], file_name) + for file_name in receive_file_list: + match_result = re.fullmatch( + tsv_row[INDEX_REGEX], file_name) # 「③」で読み込んだファイルに記載されている全てが「④１．」で取得したリストに存在した場合 if match_result is not None: # 存在したファイルの年月部分を抜き出し、チェック対象年月（処理稼働月-1)である場合 - if True: # TODO - logger.info(f'I-04-04 I/Fファイルの受領を確認しました　ファイル名：{file_name}') + match_group_yyyymm = match_result.group(1) + check_target_yyyymm = (datetime.datetime.now( + ZoneInfo(TZ)) - relativedelta(month=1)).strftime('%Y%m') + if match_group_yyyymm == check_target_yyyymm: + logger.info( + f'I-04-04 I/Fファイルの受領を確認しました　ファイル名：{file_name}') is_file_not_exists = False match_count += 1 else: - logger.info(f'I-04-07 I/Fファイルの年月がチェック対象年月と一致しません　ファイル名：{file_name}') + logger.info( + f'I-04-07 I/Fファイルの年月がチェック対象年月と一致しません　ファイル名：{file_name}') mail_message += f'{MAIL_INDENT}{tsv_row[INDEX_DATA_NAME]}(受領年月が不正:{file_name})\n' break - if is_file_not_exists: - logger.info(f'E-04-06 月次I/Fファイルに不足があります　ファイル名：{tsv_row[INDEX_DATA_NAME]}') - mail_message += f'{MAIL_INDENT}{tsv_row[INDEX_DATA_NAME]}\n' + if is_file_not_exists: + logger.info( + f'E-04-06 月次I/Fファイルに不足があります　ファイル名：{tsv_row[INDEX_DATA_NAME]}') + mail_message += f'{MAIL_INDENT}{tsv_row[INDEX_DATA_NAME]}\n' row_count += 1 @@ -234,12 +250,14 @@ def encise_data_unreceive_check(records: list, execute_month: str) -> tuple[list logger.info('I-05-01 メール送信処理開始') if len(mail_message) == 0: - logger.info(f'I-05-09 {execute_month} {receive_timing}データI/Fファイルに不足が無いため、メール送信処理をスキップします') + logger.info( + f'I-05-09 {execute_month} {receive_timing}データI/Fファイルに不足が無いため、メール送信処理をスキップします') batch_item_success.append(message_id) continue # 1.存在した場合 - logger.info(f'I-05-02 {execute_month} {receive_timing}データI/Fファイルに不足があるため、メール送信処理を開始します') + logger.info( + f'I-05-02 {execute_month} {receive_timing}データI/Fファイルに不足があるため、メール送信処理を開始します') try: logger.info( @@ -250,14 +268,18 @@ def encise_data_unreceive_check(records: list, execute_month: str) -> tuple[list Bucket=CONFIG_BUCKET_NAME, Key=f'{MAIL_TEMPLATE_FOLDER_PATH}/{event_parameter["notice_mail_title_template"]}' ) - mail_title_template = mail_title_response['Body'].read().decode('utf-8') + mail_title_template = ( + mail_title_response['Body'].read().decode('utf-8')) # 改行を取り除く - mail_title = substitute_mail_template(mail_title_template, receive_timing, mail_message) + mail_title = substitute_mail_template( + mail_title_template, receive_timing, mail_message) mail_title_without_line_break = mail_title.splitlines()[0] logger.info('I-05-04 通知メール（タイトル）テンプレートファイルを読み込みました') except Exception as e: - logger.exception(f'E-05-01 通知メール（タイトル）テンプレートファイルの読み込みに失敗しました　エラー内容：{e}') - batch_item_failures.append(make_failure_item_on_error(message_id)) + logger.exception( + f'E-05-01 通知メール（タイトル）テンプレートファイルの読み込みに失敗しました　エラー内容：{e}') + batch_item_failures.append( + make_failure_item_on_error(message_id)) continue try: @@ -269,13 +291,17 @@ def encise_data_unreceive_check(records: list, execute_month: str) -> tuple[list Bucket=CONFIG_BUCKET_NAME, Key=f'{MAIL_TEMPLATE_FOLDER_PATH}/{event_parameter["notice_mail_body_template"]}' ) - mail_body_template = mail_body_template_response['Body'].read().decode('utf-8') + mail_body_template = mail_body_template_response['Body'].read().decode( + 'utf-8') # メール本文内のプレースホルダーを置き換える - mail_body = substitute_mail_template(mail_body_template, receive_timing, mail_message) + mail_body = substitute_mail_template( + mail_body_template, receive_timing, mail_message) logger.info('I-05-06 通知メール（本文）テンプレートファイルを読み込みました') except Exception as e: - logger.exception(f'E-05-02 通知メール（本文）テンプレートファイルの読み込みに失敗しました　エラー内容：{e}') - batch_item_failures.append(make_failure_item_on_error(message_id)) + logger.exception( + f'E-05-02 通知メール（本文）テンプレートファイルの読み込みに失敗しました　エラー内容：{e}') + batch_item_failures.append( + make_failure_item_on_error(message_id)) continue logger.info(f'I-05-07 メール送信指示をします　送信先トピック：{MBJ_NOTICE_TOPIC}') @@ -310,7 +336,8 @@ def lambda_handler(event, context): batch_item_success = [] # ② SQSメッセージ重複排除処理を行う logger.info('I-02-01 メッセージ処理開始') - batch_item_failures, batch_item_success = encise_data_unreceive_check(event["Records"], execute_month) + batch_item_failures, batch_item_success = encise_data_unreceive_check( + event["Records"], execute_month) logger.info('I-06-01 すべてのメッセージの処理完了') # ⑦ メッセージを処理済として、以下のDynamoDBテーブルに記録する