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 2024-02-13 16:24:16 +09:00
parent 1153cfa8b5
commit f1ce484d1d
1 changed files with 59 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

91
lambda/encise-data-unreceive-check/encise-data-unreceive-check.py
View File

@ -8,6 +8,7 @@ import re
from zoneinfo import ZoneInfo from zoneinfo import ZoneInfo
import boto3 import boto3
from dateutil.relativedelta import relativedelta
# 環境変数 # 環境変数
CHECK_BUCKET_NAME = os.environ["CHECK_BUCKET_NAME"] 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"] MAIL_TEMPLATE_FOLDER_PATH = os.environ["MAIL_TEMPLATE_FOLDER_PATH"]
MBJ_NOTICE_TOPIC = os.environ["MBJ_NOTICE_TOPIC"] MBJ_NOTICE_TOPIC = os.environ["MBJ_NOTICE_TOPIC"]
PROCESSED_MESSAGE_DYNAMODB_TABLE_NAME = os.environ["PROCESSED_MESSAGE_DYNAMODB_TABLE_NAME"] 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"] 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)) 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() record_expiration_time = record_expiration_datetime.timestamp()
for message_id in batch_item_success: 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"] message_id = record["messageId"]
# 2.DynamoDBテーブルからレコードを取得し、処理済みメッセージかどうかを判別する # 2.DynamoDBテーブルからレコードを取得し、処理済みメッセージかどうかを判別する
if is_duplicate_message(message_id): if is_duplicate_message(message_id):
logger.info(f'受信したメッセージは既に処理済みのため、処理をスキップします。メッセージID: {message_id}') logger.info(
f'受信したメッセージは既に処理済みのため、処理をスキップします。メッセージID: {message_id}')
continue continue
except Exception as e: except Exception as e:
logger.exception(f"E-02-01 メッセージ重複チェック処理に失敗しました　エラー内容：{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 continue
# SQSパラメータをJSONシリアライズし、Pythonの辞書オブジェクト(イベントパラメータ)を取得する。 # SQSパラメータをJSONシリアライズし、Pythonの辞書オブジェクト(イベントパラメータ)を取得する。
@ -181,49 +186,60 @@ def encise_data_unreceive_check(records: list, execute_month: str) -> tuple[list
) )
logger.info('I-03-02 受領チェック対象ファイルリストを読み込みました') logger.info('I-03-02 受領チェック対象ファイルリストを読み込みました')
except Exception as e: except Exception as e:
logger.exception(f"E-03-01 受領チェック対象ファイルリストの読み込みに失敗しました　エラー内容：{e}") logger.exception(
batch_item_failures.append(make_failure_item_on_error(message_id)) f"E-03-01 受領チェック対象ファイルリストの読み込みに失敗しました　エラー内容：{e}")
batch_item_failures.append(
make_failure_item_on_error(message_id))
continue continue
# ④ 受領チェック処理を行う # ④ 受領チェック処理を行う
receive_timing = event_parameter['receive_timing'] receive_timing = event_parameter['receive_timing']
logger.info(f'I-04-01 Enciseデータ受領チェック ({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データバックアップ保管バケットの処理稼働月に該当するサブフォルダにあるファイル一覧を取得する # 1.Enciseデータバックアップ保管バケットの処理稼働月に該当するサブフォルダにあるファイル一覧を取得する
logger.info( logger.info(
f'I-04-02 オブジェクトリストの取得　取得先：{CHECK_BUCKET_NAME}/{object_prefix}' 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) receive_file_list_response = s3_client.list_objects_v2(
check_target_file_list = [] Bucket=CHECK_BUCKET_NAME, Prefix=object_prefix)
for content in check_target_file_list_response[['Contents']]: receive_file_list = []
for content in receive_file_list_response.get('Contents', []):
# オブジェクトのキーからファイル名を切り出してリストに追加 # オブジェクトのキーからファイル名を切り出してリストに追加
obj_key = content['Key'].rsplit('/', INDEX_SPLIT_NUM) 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ファイルチェック処理 # 2.I/Fファイルチェック処理
logger.info(f'I-04-03 Enciseデータ({receive_timing}) ファイルチェック処理開始') 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 match_count = 0
row_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 is_file_not_exists = True
for file_name in check_target_file_list: for file_name in receive_file_list:
match_result = re.fullmatch(tsv_row[INDEX_REGEX], file_name) match_result = re.fullmatch(
tsv_row[INDEX_REGEX], file_name)
# 「③」で読み込んだファイルに記載されている全てが「④１．」で取得したリストに存在した場合 # 「③」で読み込んだファイルに記載されている全てが「④１．」で取得したリストに存在した場合
if match_result is not None: if match_result is not None:
# 存在したファイルの年月部分を抜き出し、チェック対象年月（処理稼働月-1)である場合 # 存在したファイルの年月部分を抜き出し、チェック対象年月（処理稼働月-1)である場合
if True: # TODO match_group_yyyymm = match_result.group(1)
logger.info(f'I-04-04 I/Fファイルの受領を確認しました　ファイル名：{file_name}') 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 is_file_not_exists = False
match_count += 1 match_count += 1
else: 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' mail_message += f'{MAIL_INDENT}{tsv_row[INDEX_DATA_NAME]}(受領年月が不正:{file_name})\n'
break break
if is_file_not_exists: if is_file_not_exists:
logger.info(f'E-04-06 月次I/Fファイルに不足があります　ファイル名：{tsv_row[INDEX_DATA_NAME]}') logger.info(
mail_message += f'{MAIL_INDENT}{tsv_row[INDEX_DATA_NAME]}\n' 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 row_count += 1
@ -234,12 +250,14 @@ def encise_data_unreceive_check(records: list, execute_month: str) -> tuple[list
logger.info('I-05-01 メール送信処理開始') logger.info('I-05-01 メール送信処理開始')
if len(mail_message) == 0: 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) batch_item_success.append(message_id)
continue continue
# 1.存在した場合 # 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: try:
logger.info( logger.info(
@ -250,14 +268,18 @@ def encise_data_unreceive_check(records: list, execute_month: str) -> tuple[list
Bucket=CONFIG_BUCKET_NAME, Bucket=CONFIG_BUCKET_NAME,
Key=f'{MAIL_TEMPLATE_FOLDER_PATH}/{event_parameter["notice_mail_title_template"]}' 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] mail_title_without_line_break = mail_title.splitlines()[0]
logger.info('I-05-04 通知メール（タイトル）テンプレートファイルを読み込みました') logger.info('I-05-04 通知メール（タイトル）テンプレートファイルを読み込みました')
except Exception as e: except Exception as e:
logger.exception(f'E-05-01 通知メール（タイトル）テンプレートファイルの読み込みに失敗しました　エラー内容：{e}') logger.exception(
batch_item_failures.append(make_failure_item_on_error(message_id)) f'E-05-01 通知メール（タイトル）テンプレートファイルの読み込みに失敗しました　エラー内容：{e}')
batch_item_failures.append(
make_failure_item_on_error(message_id))
continue continue
try: try:
@ -269,13 +291,17 @@ def encise_data_unreceive_check(records: list, execute_month: str) -> tuple[list
Bucket=CONFIG_BUCKET_NAME, Bucket=CONFIG_BUCKET_NAME,
Key=f'{MAIL_TEMPLATE_FOLDER_PATH}/{event_parameter["notice_mail_body_template"]}' 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 通知メール（本文）テンプレートファイルを読み込みました') logger.info('I-05-06 通知メール（本文）テンプレートファイルを読み込みました')
except Exception as e: except Exception as e:
logger.exception(f'E-05-02 通知メール（本文）テンプレートファイルの読み込みに失敗しました　エラー内容：{e}') logger.exception(
batch_item_failures.append(make_failure_item_on_error(message_id)) f'E-05-02 通知メール（本文）テンプレートファイルの読み込みに失敗しました　エラー内容：{e}')
batch_item_failures.append(
make_failure_item_on_error(message_id))
continue continue
logger.info(f'I-05-07 メール送信指示をします　送信先トピック：{MBJ_NOTICE_TOPIC}') logger.info(f'I-05-07 メール送信指示をします　送信先トピック：{MBJ_NOTICE_TOPIC}')
@ -310,7 +336,8 @@ def lambda_handler(event, context):
batch_item_success = [] batch_item_success = []
# ② SQSメッセージ重複排除処理を行う # ② SQSメッセージ重複排除処理を行う
logger.info('I-02-01 メッセージ処理開始') 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 すべてのメッセージの処理完了') logger.info('I-06-01 すべてのメッセージの処理完了')
# ⑦ メッセージを処理済として、以下のDynamoDBテーブルに記録する # ⑦ メッセージを処理済として、以下のDynamoDBテーブルに記録する