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

仮完成

Browse Source
This commit is contained in:
野間 2023-07-04 15:24:05 +09:00
parent 89df6350ee
commit cca854be17
6 changed files with 318 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
ecs/jskult-batch-daily/.env.example
View File

@ -11,3 +11,6 @@ ULTMARC_BACKUP_FOLDER=ultmarc
JSKULT_CONFIG_BUCKET=**********************
JSKULT_CONFIG_CALENDAR_FOLDER=jskult/calendar
JSKULT_CONFIG_CALENDAR_HOLIDAY_LIST_FILE_NAME=jskult_holiday_list.txt
VJSK_BACKUP_FOLDER=vjsk
VJSK_DATA_SEND_FOLDER=send
VJSK_DATA_BUCKET=*************

26
ecs/jskult-batch-daily/src/aws/s3.py
View File

@ -76,6 +76,28 @@ class UltmarcBucket(S3Bucket):
self._s3_client.delete_file(self._bucket_name, dat_file_key)
class VjskBucket(S3Bucket):
_bucket_name = environment.VJSK_DATA_BUCKET
_folder = environment.VJSK_DATA_SEND_FOLDER
def list_dat_file(self):
return self._s3_client.list_objects(self._bucket_name, self._folder)
def upload_dat_file(self, vjsk_create_csv: str, csv_file_path: str):
# S3バケットにファイルを移動
csv_file_name = f'{self._folder}/{vjsk_create_csv}'
s3_client = S3Client()
s3_client.upload_file(csv_file_path, self._bucket_name, csv_file_name)
return
def backup_dat_file(self, dat_file_key: str, datetime_key: str):
# バックアップバケットにコピー
vjsk_backup_bucket = VjskBackupBucket()
dat_key = f'{self._folder}/{dat_file_key}'
backup_key = f'{vjsk_backup_bucket._folder}/{self._folder}/{datetime_key}/{dat_file_key.replace(f"{self._folder}/", "")}'
self._s3_client.copy(self._bucket_name, dat_key, vjsk_backup_bucket._bucket_name, backup_key)
class ConfigBucket(S3Bucket):
_bucket_name = environment.JSKULT_CONFIG_BUCKET
@ -96,3 +118,7 @@ class JskUltBackupBucket(S3Bucket):
class UltmarcBackupBucket(JskUltBackupBucket):
_folder = environment.ULTMARC_BACKUP_FOLDER
class VjskBackupBucket(JskUltBackupBucket):
_folder = environment.VJSK_BACKUP_FOLDER

270
ecs/jskult-batch-daily/src/batch/ultmarc/export_vjsk_csv_process.py Normal file
View File

@ -0,0 +1,270 @@
"""アルトマークデータ処理"""
from src.aws.s3 import UltmarcBucket, VjskBucket
from src.batch.common.batch_context import BatchContext
from src.db.database import Database
from src.logging.get_logger import get_logger
import tempfile
import os.path as path
import csv
logger = get_logger('V実用消化施設データ作成処理')
ultmarc_bucket = UltmarcBucket()
batch_context = BatchContext.get_instance()
sql_err_msg = "SQL実行エラーです。"
vjsk_csv_file_name = 'ComInst.csv'
def exec():
db = Database.get_instance()
try:
logger.info('バッチ処理を開始しました。')
try:
# DB接続
db.connect()
# ファイル単位でトランザクションを行う
db.begin()
except Exception as e:
logger.info('DB接続エラーです。')
raise e
# CSVファイルの作成用のSQL実行(施設)
record_inst_csv = csv_data_inst_select(db)
# CSVファイルの作成用のSQL実行(薬局)
record_pharm_csv = csv_data_pharm_select(db)
# CSVファイル作成
csv_file_path = make_csv_data(record_inst_csv, record_pharm_csv)
vjsk_bucket = VjskBucket()
try:
# s3へデータ移動
vjsk_bucket.upload_dat_file(vjsk_csv_file_name, csv_file_path)
except Exception as e:
logger.info('S3バケットDWHへCSVデータを作成できませんでした。')
raise e
try:
# 処理後ファイルをバックアップ
batch_context = BatchContext.get_instance()
vjsk_bucket.backup_dat_file(vjsk_csv_file_name, batch_context.syor_date)
except Exception as e:
logger.info('バックアップバケットへCSVデータをコピーできませんでした。')
raise e
csv_count = len(record_inst_csv) + len(record_pharm_csv)
logger.info(f'CSV出力件数: {csv_count}。')
logger.info('バッチ処理を正常に終了しました。')
except Exception as e:
raise e
finally:
# 終了時に必ずコミットする
db.commit()
db.disconnect()
return
def csv_data_inst_select(db):
# CSVファイルの作成用のSQL実行(施設)
try:
# 施設テーブル検索SQL
sql = """\
SELECT dcf_dsf_inst_cd,
inst_div_cd,
addr_unknown_reason_cd,
form_inst_name_kana,
inst_name_kana,
form_inst_name_kanji,
inst_name_kanji,
rltd_univ_prnt_cd,
bed_num,
close_flg,
estab_sche_flg,
close_start_ym,
estab_sche_ym,
ward_abolish_flg,
inst_repre_cd,
inst_repre_kana,
inst_repre,
phone_number_non_flg,
unconf_flg,
inst_phone_number,
inst_addr_kana,
inst_addr,
postal_number,
village_cd,
prefc_cd,
city_cd,
addr_display_number,
addr_cnt_kana,
addr_cnt,
manage_cd,
delete_sche_reason_cd,
hp_assrt_cd,
dup_opp_cd,
insp_item_micrb,
insp_item_serum,
insp_item_blood,
insp_item_patho,
insp_item_paras,
insp_item_biochem,
insp_item_ri,
re_exam_cd,
prmit_bed_num_other,
prmit_bed_num_mental,
prmit_bed_num_tuber,
prmit_bed_num_infection,
prmit_bed_num_sum,
prmit_bed_num_gen,
prmit_bed_num_rcup,
prmit_bed_maint_ymd,
inst_pharm_div,
abolish_ymd,
delete_flg,
filler_1,
filler_2,
filler_3,
filler_4,
filler_5,
regist_date,
create_user,
update_date,
update_user,
sys_regist_date,
regist_prgm_id,
sys_update_date,
update_prgm_id
FROM src05.com_inst ORDER BY dcf_dsf_inst_cd
"""
return db.execute_select(sql)
except Exception as e:
logger.debug(f'{sql_err_msg}')
raise e
def csv_data_pharm_select(db):
# CSVファイルの作成用のSQL実行(薬局)
try:
# 薬局テーブル検索SQL
sql = """\
SELECT dcf_dsf_inst_cd,
inst_div_cd,
addr_unknown_reason_cd,
form_inst_name_kana,
inst_name_kana,
form_inst_name_kanji,
inst_name_kanji,
'',
'',
close_flg,
estab_sche_flg,
close_start_ym,
estab_sche_ym,
'',
'',
inst_repre_kana,
inst_repre,
phone_number_non_flg,
unconf_flg,
inst_phone_number,
inst_addr_kana,
inst_addr,
postal_number,
village_cd,
prefc_cd,
city_cd,
addr_display_number,
addr_cnt_kana,
addr_cnt,
manage_cd,
delete_sche_reason_cd,
'',
dup_opp_cd,
'',
'',
'',
'',
'',
'',
'',
'',
'',
'',
'',
'',
'',
'',
'',
'',
inst_pharm_div,
abolish_ymd,
delete_flg,
filler_1,
filler_2,
filler_3,
filler_4,
filler_5,
regist_date,
create_user,
update_date,
update_user,
sys_regist_date,
regist_prgm_id,
sys_update_date,
update_prgm_id
FROM src05.com_pharm ORDER BY dcf_dsf_inst_cd
"""
return db.execute_select(sql)
except Exception as e:
logger.debug(f'{sql_err_msg}')
raise e
def make_csv_data(record_inst_csv: list, record_pharm_cs: list):
# 一時ファイルとして保存する(CSVファイル)
try:
temporary_dir = tempfile.mkdtemp()
csv_file_path = path.join(temporary_dir, vjsk_csv_file_name)
head_str = ['DCF_DSF_INST_CD', 'INST_DIV_CD', 'ADDR_UNKNOWN_REASON_CD', 'FORM_INST_NAME_KANA', 'INST_NAME_KANA',
'FORM_INST_NAME_KANJI', 'INST_NAME_KANJI', 'RLTD_UNIV_PRNT_CD', 'BED_NUM', 'CLOSE_FLG', 'ESTAB_SCHE_FLG',
'CLOSE_START_YM', 'ESTAB_SCHE_YM', 'WARD_ABOLISH_FLG', 'INST_REPRE_CD', 'INST_REPRE_KANA', 'INST_REPRE',
'PHONE_NUMBER_NON_FLG', 'UNCONF_FLG', 'INST_PHONE_NUMBER', 'INST_ADDR_KANA', 'INST_ADDR', 'POSTAL_NUMBER',
'VILLAGE_CD', 'PREFC_CD', 'CITY_CD', 'ADDR_DISPLAY_NUMBER', 'ADDR_CNT_KANA', 'ADDR_CNT', 'MANAGE_CD',
'DELETE_SCHE_REASON_CD', 'HP_ASSRT_CD', 'DUP_OPP_CD', 'INSP_ITEM_MICRB', 'INSP_ITEM_SERUM', 'INSP_ITEM_BLOOD',
'INSP_ITEM_PATHO', 'INSP_ITEM_PARAS', 'INSP_ITEM_BIOCHEM', 'INSP_ITEM_RI', 'RE_EXAM_CD', 'PRMIT_BED_NUM_OTHER',
'PRMIT_BED_NUM_MENTAL', 'PRMIT_BED_NUM_TUBER', 'PRMIT_BED_NUM_INFECTION', 'PRMIT_BED_NUM_SUM', 'PRMIT_BED_NUM_GEN',
'PRMIT_BED_NUM_RCUP', 'PRMIT_BED_MAINT_YMD', 'INST_PHARM_DIV', 'ABOLISH_YMD', 'DELETE_FLG', 'FILLER_1', 'FILLER_2',
'FILLER_3', 'FILLER_4', 'FILLER_5', 'REGIST_DATE', 'CREATE_USER', 'UPDATE_DATE', 'UPDATE_USER', 'SYS_REGIST_DATE',
'REGIST_PRGM_ID', 'SYS_UPDATE_DATE', 'UPDATE_PRGM_ID']
# Shift-JIS、CRLF、価囲いありで書き込む
with open(csv_file_path, mode='w', encoding='cp932') as csv_file:
writer = csv.writer(csv_file, delimiter=',', lineterminator='\n',
quotechar='"', doublequote=True, quoting=csv.QUOTE_ALL,
strict=True
)
# ヘッダ行書き込み
writer.writerow(head_str)
# データ部分書き込み(施設)
for record_inst_data in record_inst_csv:
record_inst_value = list(record_inst_data.values())
csv_data = ['' if n is None else n for n in record_inst_value]
writer.writerow(csv_data)
# データ部分書き込み(薬局)
for record_pharm_data in record_pharm_cs:
record_pharm_value = list(record_pharm_data.values())
csv_data = ['' if n is None else n for n in record_pharm_value]
writer.writerow(csv_data)
except Exception as e:
logger.info('ワークデータの作成に失敗しました。')
logger.info('バッチ処理を異常終了しました。')
raise e
return csv_file_path

13
ecs/jskult-batch-daily/src/batch/ultmarc/ultmarc_process.py
View File

@ -5,6 +5,7 @@ from datetime import datetime
from src.aws.s3 import UltmarcBucket
from src.batch.common.batch_context import BatchContext
from src.batch.ultmarc.datfile import DatFile
from src.batch.ultmarc import export_vjsk_csv_process
from src.batch.ultmarc.utmp_tables.ultmarc_table_mapper_factory import \
UltmarcTableMapperFactory
from src.db.database import Database
@ -61,11 +62,19 @@ def exec_import():
def exec_export():
"""V実消化用施設・薬局薬店データ作成処理"""
"""V実消化用施設データ作成処理"""
if not batch_context.is_ultmarc_imported:
logger.info('アルトマーク取込が行われていないため、V実消化用施設・薬局薬店データ作成処理をスキップします。')
logger.info('アルトマーク取込が行われていないため、V実消化用施設データ作成処理をスキップします。')
return
try:
logger.info('V実用消化施設データ作成処理: 開始')
export_vjsk_csv_process.exec()
logger.info('V実用消化施設データ作成処理: 終了')
except Exception as e:
raise BatchOperationException(e)
return
def _import_to_ultmarc_table(dat_file: DatFile):
db = Database.get_instance()

8
ecs/jskult-batch-daily/src/jobctrl_daily.py
View File

@ -75,11 +75,13 @@ def exec():
logger.debug(f'{"アルトマーク取込が行われました。" if batch_context.is_ultmarc_imported else "アルトマーク取込が行われませんでした。"}')
try:
logger.info('V実消化用施設・薬局薬店データ作成処理：起動')
logger.info('V実消化用施設データ作成処理：起動')
# ***********************実行させるためにtrueにしておく（後で消す）***********************
batch_context.is_ultmarc_imported = True
ultmarc_process.exec_export()
logger.info('V実消化用施設・薬局薬店データ作成処理：終了')
logger.info('V実消化用施設データ作成処理：終了')
except BatchOperationException as e:
logger.exception(f'V実消化用施設・薬局薬店データ作成処理エラー（異常終了）{e}')
logger.exception(f'V実消化用施設データ作成処理エラー（異常終了）{e}')
return constants.BATCH_EXIT_CODE_SUCCESS
logger.info('日次処理（V実消化）')

3
ecs/jskult-batch-daily/src/system_var/environment.py
View File

@ -15,6 +15,9 @@ ULTMARC_BACKUP_FOLDER = os.environ['ULTMARC_BACKUP_FOLDER']
JSKULT_CONFIG_BUCKET = os.environ['JSKULT_CONFIG_BUCKET']
JSKULT_CONFIG_CALENDAR_FOLDER = os.environ['JSKULT_CONFIG_CALENDAR_FOLDER']
JSKULT_CONFIG_CALENDAR_HOLIDAY_LIST_FILE_NAME = os.environ['JSKULT_CONFIG_CALENDAR_HOLIDAY_LIST_FILE_NAME']
VJSK_BACKUP_FOLDER = os.environ['VJSK_BACKUP_FOLDER']
VJSK_DATA_SEND_FOLDER = os.environ['VJSK_DATA_SEND_FOLDER']
VJSK_DATA_BUCKET = os.environ['VJSK_DATA_BUCKET']
# 初期値がある環境変数
LOG_LEVEL = os.environ.get('LOG_LEVEL', 'INFO')