とりあえずエラーなしで通せる程度の仮実装完了

ecs/jskult-batch-daily/src/batch/vjsk/vjsk_data_load_manager.py

@@ -14,7 +14,7 @@ class JjskDataLoadManager:
     def _import_to_db(src_file_name: str, condkey: str):
         db = Database.get_instance()
         table_name_org = mapper.get_org_table(condkey)
-        table_name_src = mapper.get_src_table(condkey)
+        upsert_sql = mapper.get_upsert_sql(condkey)
 
         try:
             db.connect()  # TODO:接続オプション local_infile = True が必要？
@@ -24,13 +24,14 @@ class JjskDataLoadManager:
             db.execute(f"TRUNCATE TABLE {table_name_org};")
 
             # orgにload　※warningは1148エラーになるらしい
-            sql = f"LOAD DATA LOCAL INFILE :src_file_name INTO TABLE {table_name_org} FIELDS TERMINATED BY '\\t' ENCLOSED BY '\"' IGNORE 1 LINES;"
+            sql = f"LOAD DATA LOCAL INFILE :src_file_name INTO TABLE {table_name_org}" \
+                " FIELDS TERMINATED BY '\\t' ENCLOSED BY '\"' IGNORE 1 LINES;"
             result = db.execute(sql, {"src_file_name": src_file_name})
             logger.info(f'tsvデータをorgテーブルにLOAD : 件数({result.rowcount})')
 
             # org→srcにinsert select
-            # TODO: INTO句とSELECT句はmapperに持たせてcondkeyで引っ張ってくるようにしたい
-            f"INSERT INTO {table_name_src} SELECT * FROM {table_name_org};"
+            result = db.execute(upsert_sql)
+            logger.info(f'orgテーブルをsrcテーブルにUPSERT : 件数({result.rowcount})')
 
             db.commit()
         except Exception as e:  # TODO:DB例外だけキャッチしたい
@@ -46,13 +47,8 @@ class JjskDataLoadManager:
         logger.debug(f'JjskDataLoadManager#load start target:{target}')
         # target : {"condkey": key, "src_file_path":local_file_path}
 
-        # データファイルオープン
+        # S3からローカルストレージにdownloadした登録対象のtsvファイルパスを取得
         local_file_name = target["src_file_path"]
-        # dat_file = VjskDatFile.retrieve_from_file(local_file_name)
-
-        # TODO: tsvファイルをload投入用のDMLに加工(システム日時つけたり、エンコードをUTF-8に変換したり)
-        # TODO: ファイルエンコード判定の参考　https://zenn.dev/takedato/articles/c3a491546f8c58
-        # TODO: エンコード変換の参考　https://dev.classmethod.jp/articles/python-encoding/
 
         # データベース登録
         self._import_to_db(local_file_name, target["condkey"])

ecs/jskult-batch-daily/src/batch/vjsk/vjsk_importer.py

@@ -114,41 +114,41 @@ def _import_file_to_db():
     if batch_context.is_import_target_vjsk_stockslipdata:
         JjskDataLoadManager.Load(target_dict[vjsk_mapper.CONDKEY_STOCK_SLIP_DATA])
 
-    # # # ファイル存在確認 卸販売データ
-    # JjskDataLoadManager.Load(target_dict[vjsk_mapper.CONDKEY_SLIP_DATA])
+    # DB登録 卸販売データ
+    JjskDataLoadManager.Load(target_dict[vjsk_mapper.CONDKEY_SLIP_DATA])
 
-    # # # ファイル存在確認 卸組織変換マスタ
-    # JjskDataLoadManager.Load(target_dict[vjsk_mapper.CONDKEY_ORG_CNV_MST])
+    # DB登録 卸組織変換マスタ
+    JjskDataLoadManager.Load(target_dict[vjsk_mapper.CONDKEY_ORG_CNV_MST])
 
-    # # # ファイル存在確認 施設統合マスタ
-    # JjskDataLoadManager.Load(target_dict[vjsk_mapper.CONDKEY_VOP_HCO_MERGE])
+    # DB登録 施設統合マスタ
+    JjskDataLoadManager.Load(target_dict[vjsk_mapper.CONDKEY_VOP_HCO_MERGE])
 
-    # # # ファイル存在確認 卸マスタ
-    # JjskDataLoadManager.Load(target_dict[vjsk_mapper.CONDKEY_WHS_MST])
+    # DB登録 卸マスタ
+    JjskDataLoadManager.Load(target_dict[vjsk_mapper.CONDKEY_WHS_MST])
 
-    # # # ファイル存在確認 卸ホールディングスマスタ
-    # JjskDataLoadManager.Load(target_dict[vjsk_mapper.CONDKEY_HLD_MST])
+    # DB登録 卸ホールディングスマスタ
+    JjskDataLoadManager.Load(target_dict[vjsk_mapper.CONDKEY_HLD_MST])
 
-    # # # ファイル存在確認 施設マスタ
-    # JjskDataLoadManager.Load(target_dict[vjsk_mapper.CONDKEY_FCL_MST])
+    # DB登録 施設マスタ
+    JjskDataLoadManager.Load(target_dict[vjsk_mapper.CONDKEY_FCL_MST])
 
-    # # # ファイル存在確認 メーカー卸組織展開表
-    # JjskDataLoadManager.Load(target_dict[vjsk_mapper.CONDKEY_MKR_ORG_HORIZON])
+    # DB登録 メーカー卸組織展開表
+    JjskDataLoadManager.Load(target_dict[vjsk_mapper.CONDKEY_MKR_ORG_HORIZON])
 
-    # # # ファイル存在確認 取引区分マスタ
-    # JjskDataLoadManager.Load(target_dict[vjsk_mapper.CONDKEY_TRAN_KBN_MST])
+    # DB登録 取引区分マスタ
+    JjskDataLoadManager.Load(target_dict[vjsk_mapper.CONDKEY_TRAN_KBN_MST])
 
-    # # # ファイル存在確認 製品マスタ
-    # JjskDataLoadManager.Load(target_dict[vjsk_mapper.CONDKEY_PHM_PRD_MST])
+    # DB登録 製品マスタ
+    JjskDataLoadManager.Load(target_dict[vjsk_mapper.CONDKEY_PHM_PRD_MST])
 
-    # # # ファイル存在確認 製品価格マスタ
-    # JjskDataLoadManager.Load(target_dict[vjsk_mapper.CONDKEY_PHM_PRICE_MST])
+    # DB登録 製品価格マスタ
+    JjskDataLoadManager.Load(target_dict[vjsk_mapper.CONDKEY_PHM_PRICE_MST])
 
-    # # # ファイル存在確認 卸得意先情報マスタ
-    # JjskDataLoadManager.Load(target_dict[vjsk_mapper.CONDKEY_WHS_CUSTOMER_MST])
+    # DB登録 卸得意先情報マスタ
+    JjskDataLoadManager.Load(target_dict[vjsk_mapper.CONDKEY_WHS_CUSTOMER_MST])
 
-    # # # ファイル存在確認 MDBコード変換マスタ
-    # JjskDataLoadManager.Load(target_dict[vjsk_mapper.CONDKEY_MDB_CONV_MST])
+    # DB登録 MDBコード変換マスタ
+    JjskDataLoadManager.Load(target_dict[vjsk_mapper.CONDKEY_MDB_CONV_MST])
 
     _logger.debug('V実消化取込処理：終了')