不要部分の削除
This commit is contained in:
mori.k
parent
e8253f27da
commit
3607229dee
@ -1,9 +1,5 @@
|
||||
import os.path as path
|
||||
import tempfile
|
||||
import shutil
|
||||
import os
|
||||
import os.path as path
|
||||
import gzip
|
||||
|
||||
import boto3
|
||||
from src.system_var import environment
|
||||
@ -141,72 +137,4 @@ class JskSendBucket(S3Bucket):
|
||||
dat_key = f'{self._send_folder}/{dat_file_key}'
|
||||
backup_key = f'{jskult_backup_bucket._folder}/{self._send_folder}/{datetime_key}/{dat_file_key.replace(f"{self._send_folder}/", "")}'
|
||||
self._s3_client.copy(self._bucket_name, dat_key,
|
||||
jskult_backup_bucket._bucket_name, backup_key)
|
||||
|
||||
class JskIOBucket(S3Bucket):
|
||||
_bucket_name = environment.JSK_IO_BUCKET
|
||||
_recv_folder = environment.JSK_RECEIVE_FOLDER
|
||||
|
||||
_s3_file_list = None
|
||||
|
||||
def get_file_list(self):
|
||||
self._s3_file_list = self._s3_client.list_objects(
|
||||
self._bucket_name, self._recv_folder)
|
||||
return self._s3_file_list
|
||||
|
||||
def download_data_file(self, data_filename: str):
|
||||
temporary_dir = tempfile.mkdtemp()
|
||||
temporary_file_path = path.join(
|
||||
temporary_dir, f'{data_filename.replace(f"{self._recv_folder}/", "")}')
|
||||
with open(temporary_file_path, mode='wb') as f:
|
||||
self._s3_client.download_file(self._bucket_name, data_filename, f)
|
||||
f.seek(0)
|
||||
return temporary_file_path
|
||||
|
||||
def unzip_data_file(self, filename: str):
|
||||
temp_dir = os.path.dirname(filename)
|
||||
decompress_filename = os.path.basename(filename).replace('.gz', '')
|
||||
decompress_file_path = os.path.join(temp_dir, decompress_filename)
|
||||
with gzip.open(filename, 'rb') as gz:
|
||||
with open(decompress_file_path, 'wb') as decompressed_file:
|
||||
shutil.copyfileobj(gz, decompressed_file)
|
||||
|
||||
ret = [decompress_file_path]
|
||||
return ret
|
||||
|
||||
def transfer_file_to_import(self, target_file: dict):
|
||||
data_import_bucket = DataImportBucket()
|
||||
transfer_from_file_path = target_file.get("filename")
|
||||
transfer_to_filename = transfer_from_file_path.replace(
|
||||
f"{self._recv_folder}/", "")
|
||||
data_import_key = f'{data_import_bucket._folder}/{transfer_to_filename}'
|
||||
self._s3_client.copy(self._bucket_name, transfer_from_file_path,
|
||||
data_import_bucket._bucket_name, data_import_key)
|
||||
|
||||
def backup_file(self, target_file: dict, datetime_key: str):
|
||||
jsk_backup_bucket = JskBackupBucket()
|
||||
backup_from_file_path = target_file.get("filename")
|
||||
backup_to_filename = backup_from_file_path.replace(
|
||||
f"{self._recv_folder}/", "")
|
||||
backup_key = f'{jsk_backup_bucket._folder}/{datetime_key}/{backup_to_filename}'
|
||||
self._s3_client.copy(self._bucket_name, backup_from_file_path,
|
||||
jsk_backup_bucket._bucket_name, backup_key)
|
||||
|
||||
def delete_file(self, target_file: dict):
|
||||
delete_path = target_file.get("filename")
|
||||
self._s3_client.delete_file(
|
||||
self._bucket_name, delete_path)
|
||||
class JskultArchiveBucket(S3Bucket):
|
||||
_bucket_name = environment.JSKULT_ARCHIVE_BUCKET
|
||||
|
||||
def upload_archive_zip_file(self, archive_zip: str, archive_zip_path: str, send_folder: str):
|
||||
# S3バケットにファイルを移動
|
||||
archive_zip_name = f'{send_folder}/{archive_zip}'
|
||||
s3_client = S3Client()
|
||||
s3_client.upload_file(
|
||||
archive_zip_path, self._bucket_name, archive_zip_name)
|
||||
return f"{self._bucket_name}/{archive_zip_name}"
|
||||
|
||||
class DataImportBucket(S3Bucket):
|
||||
_bucket_name = environment.DATA_IMPORT_BUCKET
|
||||
_folder = environment.DATA_IMPORT_FOLDER
|
||||
jskult_backup_bucket._bucket_name, backup_key)
|
||||
@ -1,75 +0,0 @@
|
||||
import csv
|
||||
import os.path as path
|
||||
import tempfile
|
||||
import zipfile
|
||||
from datetime import timedelta
|
||||
|
||||
from src.aws.s3 import JskultArchiveBucket
|
||||
from src.manager.jskult_archive_manager import JskultArchiveManager
|
||||
from src.logging.get_logger import get_logger
|
||||
|
||||
logger = get_logger("実消化_過去データアーカイブ処理")
|
||||
|
||||
|
||||
def exec():
|
||||
"""実消化_過去データアーカイブ処理"""
|
||||
try:
|
||||
logger.info("処理開始:実消化_過去データアーカイブ処理")
|
||||
jskult_archive_manager = JskultArchiveManager()
|
||||
# アーカイブ管理テーブルから対象テーブル、条件項目、条件年月、実行間隔（月）、前回条件年月、保存先を取得
|
||||
jskult_archive_manage_data_list = jskult_archive_manager.get_archive_manage()
|
||||
|
||||
# 取得したレコード分繰り返す
|
||||
for jskult_archive_manage_data in jskult_archive_manage_data_list:
|
||||
# 対象テーブルで条件項目が条件年月以前のデータを取得
|
||||
archive_data = jskult_archive_manager.get_archive_data(
|
||||
jskult_archive_manage_data["target_table"], jskult_archive_manage_data["filter_column"], jskult_archive_manage_data["filter_date"])
|
||||
# 取得データが0件の場合、スキップする
|
||||
if not archive_data:
|
||||
logger.info(
|
||||
f"アーカイブ対象データがありませんでした。対象テーブル：{jskult_archive_manage_data['target_table']} 条件年月：{jskult_archive_manage_data['filter_date']}")
|
||||
continue
|
||||
|
||||
# 一時フォルダ作成
|
||||
with tempfile.TemporaryDirectory() as temporary_dir:
|
||||
# 取得したデータをCSVに出力
|
||||
day_after_prev_filter_date = jskult_archive_manage_data["prev_filter_date"] + timedelta(
|
||||
days=1)
|
||||
file_name = f'{jskult_archive_manage_data["target_table"]}_{day_after_prev_filter_date.strftime('%Y%m%d')}_{jskult_archive_manage_data["filter_date"].strftime('%Y%m%d')}'
|
||||
csv_file_path = path.join(temporary_dir, f"{file_name}.csv")
|
||||
headers = archive_data[0].keys()
|
||||
with open(csv_file_path, 'w', newline='') as file:
|
||||
writer = csv.DictWriter(
|
||||
file, fieldnames=headers, quoting=csv.QUOTE_ALL)
|
||||
writer.writeheader()
|
||||
writer.writerows(archive_data)
|
||||
logger.info(f"CSVファイル作成に成功しました。{file_name}.csv")
|
||||
|
||||
# 作成したCSVをzip形式に圧縮
|
||||
zip_file_path = path.join(temporary_dir, f"{file_name}.zip")
|
||||
with zipfile.ZipFile(zip_file_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
|
||||
zipf.write(csv_file_path)
|
||||
logger.info(f"zip形式への圧縮に成功しました。{file_name}.zip")
|
||||
|
||||
# 圧縮したCSVを保存先へアップロード
|
||||
archive_bucket = JskultArchiveBucket()
|
||||
upload_file_path = archive_bucket.upload_archive_zip_file(
|
||||
f"{file_name}.zip", zip_file_path, jskult_archive_manage_data["archive_storage"])
|
||||
logger.info(f"{upload_file_path}へのアップロードに成功しました。")
|
||||
|
||||
# アーカイブしたデータをDBから削除
|
||||
jskult_archive_manager.delete_archive_data(
|
||||
jskult_archive_manage_data["target_table"],
|
||||
jskult_archive_manage_data["filter_column"],
|
||||
jskult_archive_manage_data["filter_date"])
|
||||
logger.info(
|
||||
f"アーカイブしたデータのDBから削除に成功しました。対象テーブル：{jskult_archive_manage_data['target_table']} 条件年月：{jskult_archive_manage_data['filter_date']}")
|
||||
|
||||
# 次回に向けてアーカイブ管理テーブルを更新する
|
||||
jskult_archive_manager.update_archive_manage(
|
||||
jskult_archive_manage_data["target_table"])
|
||||
logger.info(
|
||||
f"アーカイブ管理テーブルの更新に成功しました。対象テーブル：{jskult_archive_manage_data['target_table']}")
|
||||
logger.info("処理終了:実消化_過去データアーカイブ処理")
|
||||
except Exception as e:
|
||||
logger.exception(f"異常終了:実消化_過去データアーカイブ処理 {e}")
|
||||
@ -15,8 +15,6 @@ class DCFInstMergeEnvironment(JskultBatchEnvironment):
|
||||
self.PROCESS_NAME = environment.PROCESS_NAME
|
||||
self.JSK_DATA_SEND_FOLDER = environment.JSK_DATA_SEND_FOLDER
|
||||
self.JSK_BACKUP_FOLDER = environment.JSK_BACKUP_FOLDER
|
||||
self.TRANSFER_RESULT_FOLDER = environment.TRANSFER_RESULT_FOLDER
|
||||
self.TRANSFER_RESULT_FILE_NAME = environment.TRANSFER_RESULT_FILE_NAME
|
||||
self.DCF_INST_MERGE_SEND_FILE_NAME = environment.DCF_INST_MERGE_SEND_FILE_NAME
|
||||
|
||||
def validate(self):
|
||||
@ -40,9 +38,5 @@ class DCFInstMergeEnvironment(JskultBatchEnvironment):
|
||||
super()._assert_variable_not_empty(
|
||||
self.JSK_DATA_SEND_FOLDER, 'JSK_DATA_SEND_FOLDER')
|
||||
super()._assert_variable_not_empty(self.JSK_BACKUP_FOLDER, 'JSK_BACKUP_FOLDER')
|
||||
super()._assert_variable_not_empty(
|
||||
self.TRANSFER_RESULT_FOLDER, 'TRANSFER_RESULT_FOLDER')
|
||||
super()._assert_variable_not_empty(
|
||||
self.TRANSFER_RESULT_FILE_NAME, 'TRANSFER_RESULT_FILE_NAME')
|
||||
super()._assert_variable_not_empty(self.DCF_INST_MERGE_SEND_FILE_NAME,
|
||||
'DCF_INST_MERGE_SEND_FILE_NAME')
|
||||
|
||||
@ -4,14 +4,10 @@ from src.batch.jskult_batch_entrypoint import JskultBatchEntrypoint
|
||||
from src.aws.s3 import (JskTransferListBucket, JskIOBucket)
|
||||
from src.batch.environment.update_business_day_environment import \
|
||||
UpdateBusinessDayEnvironment
|
||||
from src.batch import archive_jsk_data
|
||||
from src.manager.jskult_batch_run_manager import JskultBatchRunManager
|
||||
from src.manager.jskult_batch_status_manager import JskultBatchStatusManager
|
||||
from src.manager.jskult_hdke_tbl_manager import JskultHdkeTblManager
|
||||
from src.manager.jskult_archive_manager import JskultArchiveManager
|
||||
from src.error.exceptions import (NotReadyException,
|
||||
EnvironmentVariableNotSetException,
|
||||
)
|
||||
from src.error.exceptions import EnvironmentVariableNotSetException
|
||||
from src.logging.get_logger import get_logger
|
||||
from src.system_var import constants
|
||||
|
||||
@ -35,8 +31,6 @@ class UpdateBusinessDay(JskultBatchEntrypoint):
|
||||
logger.info('I-1 処理開始: 実消化＆アルトマーク_日付テーブル更新')
|
||||
|
||||
jskult_hdke_tbl_manager = JskultHdkeTblManager()
|
||||
jskult_archive_manager = JskultArchiveManager(
|
||||
)
|
||||
|
||||
jskult_batch_status_manager = JskultBatchStatusManager(
|
||||
self.environment.PROCESS_NAME,
|
||||
@ -58,7 +52,6 @@ class UpdateBusinessDay(JskultBatchEntrypoint):
|
||||
'日次バッチ処理中でない、もしくはdump取得が正常終了していないため、日次バッチ処理を異常終了します。')
|
||||
# バッチ実行管理テーブルをfailedで登録
|
||||
jskult_batch_run_manager.batch_failed()
|
||||
raise NotReadyException()
|
||||
|
||||
if not jskult_batch_status_manager.can_run_business_day_update():
|
||||
# 後続処理の起動条件を満たしていない場合
|
||||
@ -73,19 +66,18 @@ class UpdateBusinessDay(JskultBatchEntrypoint):
|
||||
|
||||
return
|
||||
|
||||
_, _, archive_date, _, _, _ = jskult_archive_manager.get_archive_manage()
|
||||
|
||||
_, _, process_date = jskult_hdke_tbl_manager.get_batch_statuses()
|
||||
|
||||
|
||||
# archive取得日calendarの取得
|
||||
get_archive_calendar = None
|
||||
try:
|
||||
if archive_date == process_date:
|
||||
if get_archive_calendar == process_date:
|
||||
logger.info('[NOTICE]実消化データアーカイブ取得処理を実行します。')
|
||||
# 処理ステータスを「処理中」に設定
|
||||
jskult_batch_status_manager.set_process_status(
|
||||
constants.PROCESS_STATUS_DOING)
|
||||
|
||||
jskult_archive_manager.get_archive_data()
|
||||
archive_jsk_data.exec()
|
||||
|
||||
dt = datetime.strptime(process_date, "%Y/%m/%d")
|
||||
# 日付テーブルの処理年月日が月曜日の場合
|
||||
|
||||
@ -16,6 +16,3 @@ class MaxRunCountReachedException(MeDaCaException):
|
||||
|
||||
class EnvironmentVariableNotSetException(MeDaCaException):
|
||||
pass
|
||||
|
||||
class NotReadyException(MeDaCaException):
|
||||
pass
|
||||
|
||||
@ -1,112 +0,0 @@
|
||||
from src.db.database import Database
|
||||
from src.logging.get_logger import get_logger
|
||||
|
||||
logger = get_logger("アーカイブ管理テーブル操作")
|
||||
|
||||
|
||||
class JskultArchiveManager:
|
||||
_db: Database = None
|
||||
|
||||
def __init__(self):
|
||||
self._db = Database.get_instance()
|
||||
|
||||
def get_archive_manage(self):
|
||||
"""対象テーブル、対象項目、対象年月、実行間隔（月）、前回対象年月、保存先を取得"""
|
||||
try:
|
||||
logger.info("処理開始：get_archive_manage")
|
||||
sql = """
|
||||
select
|
||||
target_table
|
||||
, filter_column
|
||||
, filter_date
|
||||
, run_interval_months
|
||||
, prev_filter_date
|
||||
, archive_storage
|
||||
from
|
||||
internal07.jskult_archive_manage;
|
||||
"""
|
||||
self._db.connect()
|
||||
jskult_archive_manage_data = self._db.execute_select(sql)
|
||||
logger.info("処理終了：get_archive_manage")
|
||||
return jskult_archive_manage_data
|
||||
except Exception as e:
|
||||
logger.info("異常終了：get_archive_manage")
|
||||
raise
|
||||
finally:
|
||||
self._db.disconnect()
|
||||
|
||||
def get_archive_data(self,target_table:str, filter_column:str, filter_date:str):
|
||||
"""アーカイブするデータを取得"""
|
||||
try:
|
||||
logger.info("処理開始：get_archive_data")
|
||||
sql = f"""
|
||||
select
|
||||
*
|
||||
from
|
||||
src07.{target_table}
|
||||
where
|
||||
str_to_date({filter_column},'%Y%m%d') <= :filter_date;
|
||||
"""
|
||||
self._db.connect()
|
||||
parameter_dict = {'filter_date' : filter_date}
|
||||
target_table_data = self._db.execute_select(sql, parameter_dict)
|
||||
logger.info("処理終了：get_archive_data")
|
||||
return target_table_data
|
||||
except Exception as e:
|
||||
logger.info("異常終了：get_archive_data")
|
||||
raise
|
||||
finally:
|
||||
self._db.disconnect()
|
||||
|
||||
def delete_archive_data(self, target_table:str, filter_column:str, filter_date:str):
|
||||
"""アーカイブしたデータを削除"""
|
||||
try:
|
||||
logger.info("処理開始：delete_archive_data")
|
||||
sql = f"""
|
||||
delete from
|
||||
src07.{target_table}
|
||||
where
|
||||
str_to_date({filter_column},'%Y%m%d') <= :filter_date;
|
||||
"""
|
||||
self._db.connect()
|
||||
self._db.begin()
|
||||
parameter_dict = {'filter_date' : filter_date}
|
||||
self._db.execute(sql, parameter_dict)
|
||||
self._db.commit()
|
||||
logger.info("処理終了：delete_archive_data")
|
||||
return
|
||||
except Exception as e:
|
||||
self._db.rollback()
|
||||
logger.info("異常終了：delete_archive_data")
|
||||
raise
|
||||
finally:
|
||||
self._db.disconnect()
|
||||
|
||||
def update_archive_manage(self, target_table:str):
|
||||
"""アーカイブ管理テーブルの指定した対象テーブルのレコードを更新する"""
|
||||
try:
|
||||
logger.info("処理開始：update_archive_manage")
|
||||
sql = f"""
|
||||
update internal07.jskult_archive_manage
|
||||
set
|
||||
prev_filter_date = filter_date
|
||||
, filter_date = LAST_DAY(
|
||||
DATE_ADD(filter_date, INTERVAL run_interval_months MONTH)
|
||||
)
|
||||
, upd_user = CURRENT_USER()
|
||||
, upd_date = NOW()
|
||||
where
|
||||
target_table = '{target_table}';
|
||||
"""
|
||||
self._db.connect()
|
||||
self._db.begin()
|
||||
self._db.execute(sql)
|
||||
self._db.commit()
|
||||
logger.info("処理終了：update_archive_manage")
|
||||
return
|
||||
except Exception as e:
|
||||
self._db.rollback()
|
||||
logger.info("異常終了：update_archive_manage")
|
||||
raise
|
||||
finally:
|
||||
self._db.disconnect()
|
||||
@ -22,12 +22,8 @@ JSKULT_BACKUP_BUCKET = os.environ.get('JSKULT_BACKUP_BUCKET', None)
|
||||
JSK_IO_BUCKET = os.environ.get('JSK_IO_BUCKET', None)
|
||||
JSK_BACKUP_FOLDER = os.environ.get('JSK_BACKUP_FOLDER', None)
|
||||
JSK_DATA_SEND_FOLDER = os.environ.get('JSK_DATA_SEND_FOLDER', None)
|
||||
JSK_RECEIVE_FOLDER = os.environ.get('JSK_RECEIVE_FOLDER', None)
|
||||
BATCH_MANAGE_DYNAMODB_TABLE_NAME = os.environ.get(
|
||||
'BATCH_MANAGE_DYNAMODB_TABLE_NAME', None)
|
||||
DATA_IMPORT_BUCKET = os.environ.get('DATA_IMPORT_BUCKET', None)
|
||||
DATA_IMPORT_FOLDER = os.environ.get('DATA_IMPORT_FOLDER', None)
|
||||
JSKULT_ARCHIVE_BUCKET = os.environ.get('JSKULT_ARCHIVE_BUCKET', None)
|
||||
|
||||
|
||||
# 初期値がある環境変数
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user