Merge pull request #210 feature-NEWDWH2021-1062 into develop

ecs/jskult-batch-daily/src/batch/dcf_inst_merge/integrate_dcf_inst_merge.py

@@ -0,0 +1,647 @@
+from datetime import datetime, timedelta
+from src.batch.batch_functions import logging_sql
+from src.batch.common.batch_context import BatchContext
+from src.db.database import Database
+from src.error.exceptions import BatchOperationException
+from src.logging.get_logger import get_logger
+from src.time.elapsed_time import ElapsedTime
+
+batch_context = BatchContext.get_instance()
+logger = get_logger('DCF施設統合マスタ日次更新')
+
+
+def exec():
+    db = Database.get_instance()
+    try:
+        db.connect()
+        db.begin()
+        logger.debug('DCF施設統合マスタ日次更新処理開始')
+        # DCF施設統合マスタ移行先コードのセット(無効フラグが『0(有効)』)
+        enabled_dst_inst_merge_records = _set_enabled_dct_inst_merge(db)
+        # DCF施設統合マスタ移行先コードのセット(無効フラグが『1(無効)』)
+        _set_disabled_dct_inst_merge(db)
+        # DCF施設統合マスタに無効フラグが『0(有効)』データが存在する場合
+        if len(enabled_dst_inst_merge_records) > 0:
+            _add_emp_chg_inst(db, enabled_dst_inst_merge_records)
+            _add_ult_ident_presc(db, enabled_dst_inst_merge_records)
+        db.commit()
+        logger.debug('DCF施設統合マスタ日次更新処理終了')
+    except Exception as e:
+        db.rollback()
+        raise BatchOperationException(e)
+    finally:
+        db.disconnect()
+
+
+def _set_enabled_dct_inst_merge(db: Database) -> list[dict]:
+    # データ取得（無効フラグが『0(有効)』)
+    enabled_dst_inst_merge_records = _select_dct_inst_merge(db, 0)
+    # 移行先DCF施設コードの更新（無効フラグが『0(有効)』)
+    if _update_dcf_inst_merge(db, 0) > 0:
+        # DCF施設統合マスタの過去分の洗い替え
+        for row in enabled_dst_inst_merge_records:
+            _update_dcf_inst_cd_new(db, row['dup_opp_cd'], row['dcf_inst_cd'], '')
+
+    return enabled_dst_inst_merge_records
+
+
+def _set_disabled_dct_inst_merge(db: Database):
+    # データ取得（無効フラグが『1(無効)』)
+    disabled_dst_inst_merge_records = _select_dct_inst_merge(db, 1)
+    # 移行先DCF施設コードの更新（無効フラグが『1(無効)』)
+    if _update_dcf_inst_merge(db, 1) > 0:
+        # DCF施設統合マスタの過去分の洗い替え
+        for row in disabled_dst_inst_merge_records:
+            _update_dcf_inst_cd_new(db, row['dcf_inst_cd'], row['dup_opp_cd'], '戻し')
+
+
+def _select_ult_ident_presc_dcf_inst_cd(db: Database, dcf_inst_cd: str) -> list[dict]:
+    # 納入先処方元マスタから、DCF施設コードに対応したレコードの取得
+    try:
+        sql = """
+            SELECT
+                ta_cd,
+                ult_ident_cd,
+                ratio
+            FROM
+                src05.ult_ident_presc
+            WHERE
+                    presc_cd = :dcf_inst_cd
+                AND (SELECT ht.syor_date FROM src05.hdke_tbl AS ht) < end_date
+        """
+        params = {'dcf_inst_cd': dcf_inst_cd}
+        ult_ident_presc_ta_cd_records = db.execute_select(sql, params)
+        logging_sql(logger, sql)
+        logger.info('納入先処方元マスタからDCF施設コードに対応したレコードの取得に成功')
+    except Exception as e:
+        logger.debug('納入先処方元マスタからDCF施設コードに対応したレコードの取得に失敗')
+        raise e
+
+    return ult_ident_presc_ta_cd_records
+
+
+def _add_ult_ident_presc(db: Database, enabled_dst_inst_merge_records: list[dict]):
+    # 納入先処方元マスタの追加
+    logger.info('納入先処方元マスタの登録 開始')
+    for data_inst_cnt, enabled_merge_record in enumerate(enabled_dst_inst_merge_records, start=1):
+        tekiyo_month_first_day = _get_first_day_of_month(enabled_merge_record['tekiyo_month'])
+        ult_ident_presc_source_records = _select_ult_ident_presc_dcf_inst_cd(db, enabled_merge_record['dcf_inst_cd'])
+        for ult_ident_presc_source_record in ult_ident_presc_source_records:
+            ult_ident_presc_records = _select_ult_ident_presc(db,
+                                                              enabled_merge_record['dcf_inst_cd'],
+                                                              enabled_merge_record['dup_opp_cd'],
+                                                              ult_ident_presc_source_record)
+            for data_cnt, ult_ident_presc_row in enumerate(ult_ident_presc_records, start=1):
+                logger.info(f'{data_inst_cnt}件目の移行施設の{data_cnt}レコード目処理 開始')
+                # 処方元コード=重複時相手先コードが発生した場合
+                if ult_ident_presc_row['opp_count'] > 0:
+                    continue
+
+                start_date = _str_to_date_time(ult_ident_presc_row['start_date'])
+                set_start_date = start_date \
+                    if start_date > tekiyo_month_first_day else tekiyo_month_first_day
+                set_start_date = _date_time_to_str(set_start_date)
+                is_exists_duplicate_key = False
+                if _count_duplicate_ult_ident_presc(db, set_start_date, ult_ident_presc_row) > 0:
+                    _delete_ult_ident_presc(db, set_start_date, ult_ident_presc_row,
+                                            '納入先処方元マスタの重複予定データの削除')
+                    is_exists_duplicate_key = True
+                else:
+                    logger.info('納入先処方元マスタの重複予定データなし')
+                _insert_ult_ident_presc(db, set_start_date, enabled_merge_record['dup_opp_cd'], ult_ident_presc_row)
+
+                # 重複予定データが存在しない、且つ、適用終了日 ≧ 適用開始日の場合
+                if not is_exists_duplicate_key and _str_to_date_time(ult_ident_presc_row['end_date']) >= start_date:
+                    last_end_date = tekiyo_month_first_day - timedelta(days=1)
+                    # 適用終了日を、DCF施設統合マスタの適用月度の前月末日で更新
+                    _update_ult_ident_presc_end_date(db, _date_time_to_str(last_end_date), ult_ident_presc_row)
+
+    logger.info('納入先処方元マスタの登録 終了')
+
+
+def _select_emp_chg_inst_ta_cd(db: Database, dcf_inst_cd: str) -> list[dict]:
+    # 従業員担当施設マスタから、DCF施設コードに対応した領域コードの取得
+    try:
+        sql = """
+            SELECT
+                ta_cd
+            FROM
+                src05.emp_chg_inst
+            WHERE
+                    inst_cd = :dcf_inst_cd
+                AND enabled_flg = 'Y'
+                AND (SELECT ht.syor_date FROM src05.hdke_tbl AS ht) < end_date
+        """
+        params = {'dcf_inst_cd': dcf_inst_cd}
+        emp_chg_inst_ta_cd_records = db.execute_select(sql, params)
+        logging_sql(logger, sql)
+        logger.info('従業員担当施設マスタから領域コードの取得に成功')
+    except Exception as e:
+        logger.debug('従業員担当施設マスタから領域コードの取得に失敗')
+        raise e
+
+    return emp_chg_inst_ta_cd_records
+
+
+def _add_emp_chg_inst(db: Database, enabled_dst_inst_merge_records: list[dict]):
+    # 従業員担当施設マスタの登録
+    logger.info('従業員担当施設マスタの登録 開始')
+    for enabled_merge_record in enabled_dst_inst_merge_records:
+        tekiyo_month_first_day = _get_first_day_of_month(enabled_merge_record['tekiyo_month'])
+        emp_chg_inst_ta_cd_records = _select_emp_chg_inst_ta_cd(db, enabled_merge_record['dcf_inst_cd'])
+        for emp_chg_inst_ta_cd_record in emp_chg_inst_ta_cd_records:
+            emp_chg_inst_records = _select_emp_chg_inst(db, enabled_merge_record['dcf_inst_cd'], enabled_merge_record['dup_opp_cd'],
+                                                        emp_chg_inst_ta_cd_record['ta_cd'])
+            for emp_chg_inst_row in emp_chg_inst_records:
+                # 重複時相手先コードが存在したかのチェック
+                if emp_chg_inst_row['opp_count'] > 0:
+                    continue
+
+                start_date = _str_to_date_time(emp_chg_inst_row['start_date'])
+                set_start_date = start_date \
+                    if start_date > tekiyo_month_first_day else tekiyo_month_first_day
+
+                _insert_emp_chg_inst(db, enabled_merge_record['dup_opp_cd'], _date_time_to_str(set_start_date),
+                                     emp_chg_inst_row)
+
+                # 適用開始日 < DCF施設統合マスタの適用月度の1日の場合
+                if start_date < tekiyo_month_first_day:
+                    # DCF施設統合マスタの適用月度の前月末日で、適用終了日を更新する
+                    last_end_date = tekiyo_month_first_day - timedelta(days=1)
+                    _update_emp_chg_inst_end_date(db, enabled_merge_record['dcf_inst_cd'], _date_time_to_str(last_end_date),
+                                                  emp_chg_inst_row)
+                    continue
+                # 適用開始日 ≧ DCF施設統合マスタの適用月度の1日の場合、N(論理削除レコード)に設定する
+                _update_emp_chg_inst_disabled(db, enabled_merge_record['dcf_inst_cd'], emp_chg_inst_row['ta_cd'],
+                                              emp_chg_inst_row['start_date'])
+
+    logger.info('従業員担当施設マスタの登録 終了')
+
+
+def _delete_ult_ident_presc(db: Database, start_date: str, ult_ident_presc_row: dict,
+                            log_message: str):
+    # ult_ident_prescのDelete
+    try:
+        elapsed_time = ElapsedTime()
+        sql = """
+            DELETE FROM
+                src05.ult_ident_presc
+            WHERE
+                    ta_cd = :ta_cd
+                AND ult_ident_cd = :ult_ident_cd
+                AND ratio = :ratio
+                AND start_date = :start_date
+        """
+        params = {
+            'ta_cd': ult_ident_presc_row['ta_cd'],
+            'ult_ident_cd': ult_ident_presc_row['ult_ident_cd'],
+            'ratio': ult_ident_presc_row['ratio'],
+            'start_date': start_date
+        }
+        res = db.execute(sql, params)
+        logging_sql(logger, sql)
+        logger.info(f'{log_message} 成功, {res.rowcount} 行更新 ({elapsed_time.of})')
+    except Exception as e:
+        logger.debug(f'{log_message} 失敗')
+        raise e
+
+
+def _update_emp_chg_inst_disabled(db: Database, dcf_inst_cd: str, ta_cd: str, start_date: str):
+    # emp_chg_instをUPDATE
+    try:
+        elapsed_time = ElapsedTime()
+        sql = """
+            UPDATE
+                src05.emp_chg_inst
+            SET
+                enabled_flg = 'N',
+                updater = CURRENT_USER(),
+                update_date = SYSDATE()
+            WHERE
+                    inst_cd = :dcf_inst_cd
+                AND ta_cd = :ta_cd
+                AND start_date = :start_date
+        """
+        params = {'dcf_inst_cd': dcf_inst_cd, 'ta_cd': ta_cd, 'start_date': start_date}
+        res = db.execute(sql, params)
+        logging_sql(logger, sql)
+        logger.info(f'従業員担当施設マスタのYorNフラグ更新に成功, {res.rowcount} 行更新 ({elapsed_time.of})')
+    except Exception as e:
+        logger.debug('従業員担当施設マスタのYorNフラグ更新に失敗')
+        raise e
+
+
+def _update_emp_chg_inst_end_date(db: Database, dcf_inst_cd: str, last_end_date: str,
+                                  emp_chg_inst_row: dict):
+    # emp_chg_instをUPDATE
+    try:
+        elapsed_time = ElapsedTime()
+        sql = """
+            UPDATE
+                src05.emp_chg_inst
+            SET end_date = :end_date,
+                updater = CURRENT_USER(),
+                update_date = SYSDATE()
+            WHERE
+                    inst_cd = :dcf_inst_cd
+                AND ta_cd = :ta_cd
+                AND emp_cd = :emp_cd
+                AND bu_cd = :bu_cd
+                AND start_date = :start_date
+        """
+        params = {
+            'end_date': last_end_date,
+            'dcf_inst_cd': dcf_inst_cd,
+            'ta_cd': emp_chg_inst_row['ta_cd'],
+            'emp_cd': emp_chg_inst_row['emp_cd'],
+            'bu_cd': emp_chg_inst_row['bu_cd'],
+            'start_date': emp_chg_inst_row['start_date']
+        }
+        res = db.execute(sql, params)
+        logging_sql(logger, sql)
+        logger.info(f'従業員担当施設マスタの適用終了日更新 成功, {res.rowcount} 行更新 ({elapsed_time.of})')
+    except Exception as e:
+        logger.debug('従業員担当施設マスタの適用終了日更新 失敗')
+        raise e
+
+
+def _insert_emp_chg_inst(db: Database, dup_opp_cd: str, set_start_date: str,
+                         emp_chg_inst_row: dict):
+    # emp_chg_instにINSERT
+    try:
+        elapsed_time = ElapsedTime()
+        sql = """
+            INSERT INTO
+                src05.emp_chg_inst(
+                    inst_cd,
+                    ta_cd,
+                    emp_cd,
+                    bu_cd,
+                    start_date,
+                    end_date,
+                    main_chg_flg,
+                    enabled_flg,
+                    creater,
+                    create_date,
+                    updater,
+                    update_date
+                )
+            VALUES(
+                :dup_opp_cd,
+                :ta_cd,
+                :emp_cd,
+                :bu_cd,
+                :start_date,
+                :end_date,
+                :main_chg_flg,
+                'Y',
+                CURRENT_USER(),
+                SYSDATE(),
+                CURRENT_USER(),
+                SYSDATE()
+            )
+        """
+        params = {
+            'dup_opp_cd': dup_opp_cd,
+            'ta_cd': emp_chg_inst_row['ta_cd'],
+            'emp_cd': emp_chg_inst_row['emp_cd'],
+            'bu_cd': emp_chg_inst_row['bu_cd'],
+            'start_date': set_start_date,
+            'end_date': emp_chg_inst_row['end_date'],
+            'main_chg_flg': None
+            if emp_chg_inst_row['main_chg_flg'] is None else emp_chg_inst_row['main_chg_flg']
+        }
+        res = db.execute(sql, params)
+        logging_sql(logger, sql)
+        logger.info(f'従業員担当施設マスタの追加に成功, {res.rowcount} 行更新 ({elapsed_time.of})')
+    except Exception as e:
+        logger.debug('従業員担当施設マスタの追加に失敗')
+        raise e
+
+
+def _select_dct_inst_merge(db: Database, muko_flg: int) -> list[dict]:
+    # dcf_inst_mergeからSELECT
+    # 無効フラグがOFFのときは、移行先DCF施設コードが設定されてないデータを抽出する。
+    # ONのときは、移行先DCF施設コードが設定されているデータを抽出する。
+    try:
+        sql = """
+            SELECT
+                dim.dcf_inst_cd,
+                dim.dup_opp_cd,
+                dim.tekiyo_month
+            FROM
+                src05.dcf_inst_merge AS dim
+            INNER JOIN
+                src05.hdke_tbl AS ht
+            ON dim.tekiyo_month = DATE_FORMAT(ht.syor_date, '%Y%m')
+            WHERE
+                    dim.muko_flg = :muko_flg
+                AND dim.enabled_flg = 'Y'
+                AND dim.dcf_inst_cd_new IS {not_null}NULL
+        """.format(
+            not_null='' if muko_flg == 0 else 'NOT '
+        )
+        params = {
+            'muko_flg': muko_flg
+        }
+        dst_inst_merge_records = db.execute_select(sql, params)
+        logging_sql(logger, sql)
+        logger.info('DCF施設統合マスタの取得に成功')
+    except Exception as e:
+        logger.debug('DCF施設統合マスタの取得に失敗')
+        raise e
+
+    return dst_inst_merge_records
+
+
+def _update_dcf_inst_merge(db: Database, muko_flg: int) -> int:
+    # dcf_inst_mergeをUPDATE
+    # 無効フラグがOFFのときは、
+    #   移行先DCF施設コードが設定されていないデータを抽出し、移行先DCF施設コードに重複時相手先コードを上書きする
+    # 無効フラグがONのときは、
+    #   移行先DCF施設コードが設定されているデータを抽出し、移行先DCF施設コードにNULLを上書きする。
+    try:
+        elapsed_time = ElapsedTime()
+        log_message = '更新しました' if muko_flg == 0 else '無効データに戻しました'
+        sql = """
+            UPDATE
+                src05.dcf_inst_merge AS updim
+                INNER JOIN(
+                    SELECT
+                        dim.dcf_inst_cd AS base_dcf_inst_cd,
+                        dim.dup_opp_cd AS base_dup_opp_cd,
+                        dim.tekiyo_month AS base_tekiyo_month,
+                        dim.muko_flg AS base_muko_flg,
+                        dim.enabled_flg AS base_enabled_flg
+                    FROM
+                        src05.dcf_inst_merge AS dim
+                    INNER JOIN
+                        src05.hdke_tbl AS ht
+                    ON dim.tekiyo_month = DATE_FORMAT(ht.syor_date, '%Y%m')
+                    WHERE
+                            dim.muko_flg = :muko_flg
+                        AND dim.enabled_flg ='Y'
+                        AND dim.dcf_inst_cd_new IS {not_null}NULL
+                ) AS bf_dim
+            SET
+                updim.dcf_inst_cd_new = {column},
+                updim.updater = CURRENT_USER(),
+                updim.update_date = SYSDATE()
+            WHERE
+                    updim.dcf_inst_cd = base_dcf_inst_cd
+                AND updim.dup_opp_cd = base_dup_opp_cd
+                AND updim.tekiyo_month = base_tekiyo_month
+                AND updim.muko_flg = base_muko_flg
+                AND updim.enabled_flg = base_enabled_flg
+        """.format(
+            not_null='' if muko_flg == 0 else 'NOT ',
+            column='base_dup_opp_cd' if muko_flg == 0 else 'NULL'
+        )
+        params = {
+            'muko_flg': muko_flg
+        }
+        res = db.execute(sql, params)
+        logging_sql(logger, sql)
+        logger.info(f'DCF施設統合マスタの有効データを{log_message} 成功, {res.rowcount} 行更新 ({elapsed_time.of})')
+    except Exception as e:
+        logger.debug(f'DCF施設統合マスタの有効データを{log_message} 失敗')
+        raise e
+
+    return res.rowcount
+
+
+def _update_dcf_inst_cd_new(db: Database, dcf_inst_cd_new_after: str, dcf_inst_cd_new_before: str, log_message: str):
+    # dcf_inst_mergeをUPDATE
+    try:
+        elapsed_time = ElapsedTime()
+        sql = """
+            UPDATE
+                src05.dcf_inst_merge
+            SET
+                dcf_inst_cd_new = :dcf_inst_cd_new_after,
+                updater = CURRENT_USER(),
+                update_date = SYSDATE()
+            WHERE
+                    dcf_inst_cd_new = :dcf_inst_cd_new_before
+                AND enabled_flg = 'Y'
+                AND muko_flg = 0
+        """
+        params = {
+            'dcf_inst_cd_new_after': dcf_inst_cd_new_after,
+            'dcf_inst_cd_new_before': dcf_inst_cd_new_before
+        }
+        res = db.execute(sql, params)
+        logging_sql(logger, sql)
+        logger.info(f'移行先DCF施設コードの{log_message}更新に成功, {res.rowcount} 行更新 ({elapsed_time.of})')
+    except Exception as e:
+        logger.debug(f'移行先DCF施設コードの{log_message}更新に失敗')
+        raise e
+
+
+def _update_ult_ident_presc_end_date(db: Database, last_end_date: str, ult_ident_presc_row: dict):
+    # ult_ident_presc_endをUPDATE
+    try:
+        elapsed_time = ElapsedTime()
+        sql = """
+            UPDATE
+                src05.ult_ident_presc
+            SET end_date = :end_date,
+                updater = CURRENT_USER(),
+                update_date = SYSDATE()
+            WHERE
+                    ta_cd = :ta_cd
+                AND ult_ident_cd = :ult_ident_cd
+                AND ratio = :ratio
+                AND start_date = :start_date
+        """
+        params = {
+            'end_date': last_end_date,
+            'ta_cd': ult_ident_presc_row['ta_cd'],
+            'ult_ident_cd': ult_ident_presc_row['ult_ident_cd'],
+            'ratio': ult_ident_presc_row['ratio'],
+            'start_date': ult_ident_presc_row['start_date']
+        }
+        res = db.execute(sql, params)
+        logging_sql(logger, sql)
+        logger.info(f'終了日 > 開始月のため適用終了日を更新 成功, {res.rowcount} 行更新 ({elapsed_time.of})')
+    except Exception as e:
+        logger.debug('終了日 > 開始月のため適用終了日を更新 失敗')
+        raise e
+
+
+def _insert_ult_ident_presc(db: Database, set_Start_Date: str, dup_opp_cd: str,
+                            ult_ident_presc_row: dict):
+    # ult_ident_prescにINSERT
+    try:
+        elapsed_time = ElapsedTime()
+        sql = """
+            INSERT INTO
+                src05.ult_ident_presc(
+                    ta_cd,
+                    ult_ident_cd,
+                    ratio,
+                    start_date,
+                    presc_cd,
+                    end_date,
+                    creater,
+                    create_date,
+                    update_date,
+                    updater
+                )
+            VALUES(
+                :ta_cd,
+                :ult_ident_cd,
+                :ratio,
+                :start_date,
+                :presc_cd,
+                :end_date,
+                CURRENT_USER(),
+                SYSDATE(),
+                SYSDATE(),
+                CURRENT_USER()
+            )
+        """
+        params = {
+            'ta_cd': ult_ident_presc_row['ta_cd'],
+            'ult_ident_cd': ult_ident_presc_row['ult_ident_cd'],
+            'ratio': ult_ident_presc_row['ratio'],
+            'start_date': set_Start_Date,
+            'presc_cd': dup_opp_cd,
+            'end_date': ult_ident_presc_row['end_date']
+        }
+        res = db.execute(sql, params)
+        logging_sql(logger, sql)
+        logger.info(f'納入先処方元マスタに追加 成功, {res.rowcount} 行更新 ({elapsed_time.of})')
+    except Exception as e:
+        logger.debug('納入先処方元マスタに追加 失敗')
+        raise e
+
+
+def _select_emp_chg_inst(db: Database, dcf_inst_cd: str, dup_opp_cd: str, ta_cd: str) -> list[dict]:
+    # emp_chg_instからSELECT
+    try:
+        sql = """
+            SELECT
+                eci.inst_cd,
+                eci.ta_cd,
+                eci.emp_cd,
+                eci.bu_cd,
+                eci.start_date,
+                eci.end_date,
+                eci.main_chg_flg,
+                eci.enabled_flg,
+                (
+                    SELECT
+                        COUNT(eciopp.inst_cd)
+                    FROM
+                        src05.emp_chg_inst AS eciopp
+                    WHERE
+                            eciopp.inst_cd = :dup_opp_cd
+                        AND eciopp.ta_cd = :ta_cd
+                ) AS opp_count
+            FROM
+                src05.emp_chg_inst AS eci
+            WHERE
+                    eci.inst_cd = :dcf_inst_cd
+                AND eci.ta_cd = :ta_cd
+                AND eci.enabled_flg = 'Y'
+                AND (SELECT ht.syor_date FROM src05.hdke_tbl AS ht) < eci.end_date
+        """
+        params = {'dcf_inst_cd': dcf_inst_cd, 'dup_opp_cd': dup_opp_cd, 'ta_cd': ta_cd}
+        emp_chg_inst_records = db.execute_select(sql, params)
+        logging_sql(logger, sql)
+        logger.info('従業員担当施設マスタの取得 成功')
+    except Exception as e:
+        logger.debug('従業員担当施設マスタの取得 失敗')
+        raise e
+    return emp_chg_inst_records
+
+
+def _select_ult_ident_presc(db: Database, dcf_inst_cd: str, dup_opp_cd: str,
+                            ult_ident_presc_row: dict) -> list[dict]:
+    # ult_ident_prescからSELECT
+    try:
+        sql = """
+            SELECT
+                uip.ta_cd,
+                uip.ult_ident_cd,
+                uip.ratio,
+                uip.start_date,
+                uip.end_date,
+                (
+                    SELECT
+                        COUNT(uipopp.ta_cd)
+                    FROM
+                        src05.ult_ident_presc AS uipopp
+                    WHERE
+                            uipopp.presc_cd = :dup_opp_cd
+                        AND uipopp.ta_cd = :ta_cd
+                        AND uipopp.ult_ident_cd = :ult_ident_cd
+                        AND uipopp.ratio = :ratio
+                ) AS opp_count
+            FROM
+                src05.ult_ident_presc AS uip
+            WHERE
+                    uip.presc_cd = :dcf_inst_cd
+                AND uip.ta_cd = :ta_cd
+                AND (SELECT ht.syor_date FROM src05.hdke_tbl AS ht) < uip.end_date
+        """
+        params = {
+            'dcf_inst_cd': dcf_inst_cd,
+            'dup_opp_cd': dup_opp_cd,
+            'ta_cd': ult_ident_presc_row['ta_cd'],
+            'ult_ident_cd': ult_ident_presc_row['ult_ident_cd'],
+            'ratio': ult_ident_presc_row['ratio']
+        }
+        ult_ident_presc_records = db.execute_select(sql, params)
+        logging_sql(logger, sql)
+        logger.info('納入先処方元マスタの取得 成功')
+    except Exception as e:
+        logger.debug('納入先処方元マスタの取得 失敗')
+        raise e
+    return ult_ident_presc_records
+
+
+def _count_duplicate_ult_ident_presc(db: Database, set_start_date: str,
+                                     ult_ident_presc_row: dict) -> int:
+    # ult_ident_prescの重複時相手先コードの件数取得
+    try:
+        sql = """
+            SELECT
+                COUNT(ta_cd) AS cnt
+            FROM
+                src05.ult_ident_presc
+            WHERE
+                    ta_cd = :ta_cd
+                AND ult_ident_cd = :ult_ident_cd
+                AND ratio = :ratio
+                AND start_date = :start_date
+        """
+        params = {
+            'ta_cd': ult_ident_presc_row['ta_cd'],
+            'ult_ident_cd': ult_ident_presc_row['ult_ident_cd'],
+            'ratio': ult_ident_presc_row['ratio'],
+            'start_date': set_start_date
+        }
+        result = db.execute_select(sql, params)
+        logging_sql(logger, sql)
+        logger.info('納入先処方元マスタの重複予定データの存在チェック 成功')
+    except Exception as e:
+        logger.debug('納入先処方元マスタの重複予定データの存在チェック 失敗')
+        raise e
+    return result[0]['cnt']
+
+
+def _get_first_day_of_month(year_month: str) -> datetime:
+    # year_monthの初日の日付を日付型に変換し返却する
+    return datetime.strptime(year_month + '01', '%Y%m%d')
+
+
+def _str_to_date_time(str_date_time: str) -> datetime:
+    # str_date_timeを日付型に変換して返却する
+    return datetime.strptime(str_date_time, '%Y%m%d')
+
+
+def _date_time_to_str(date_time: datetime) -> str:
+    # date_timeをYmd型に変換して返却する
+    return date_time.strftime('%Y%m%d')

ecs/jskult-batch-daily/src/batch/laundering/sales_laundering.py

@@ -1,5 +1,6 @@
 from src.batch.common.batch_context import BatchContext
 from src.batch.laundering import create_inst_merge_for_laundering, emp_chg_inst_laundering, ult_ident_presc_laundering
+from src.batch.dcf_inst_merge import integrate_dcf_inst_merge
 from src.logging.get_logger import get_logger
 
 batch_context = BatchContext.get_instance()
@@ -16,6 +17,8 @@ def exec():
         return
     # 洗替用マスタ作成
     create_inst_merge_for_laundering.exec()
+    # DCF施設統合マスタ日次更新
+    integrate_dcf_inst_merge.exec()
     # 施設担当者洗替
     emp_chg_inst_laundering.exec()
     # 納入先処方元マスタ洗替