import os import logging import pandas as pd import pymysql from typing import Dict, List, Optional, Tuple, Any from datetime import datetime import json from conversion_processor import ConversionProcessor import config # 初期化（コード変換表はオプション） processor = ConversionProcessor( config.DONOR_KIDNEY_RULES, config.CODE_MAPPING_EXCEL # オプション ) # ログ設定（環境変数でデバッグモードを制御） log_level = logging.DEBUG if os.getenv('DEBUG') == '1' else logging.INFO logging.basicConfig(level=log_level, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) ########################################################################################## ### 定数定義 ########################################################################################## # データベース接続設定 DB_CONFIG = { "host": "db", "user": "root", "password": "123456", "database": "dev_tracer_db2", "charset": "utf8mb4", "cursorclass": pymysql.cursors.DictCursor } # テーブル削除順序（外部キー制約を考慮） ORDERED_TABLES = [ 'T_NYURYOKUJOKYO_LIV', # 追加（他のテーブルから参照されている可能性） 'T_IJI_MENEKI_YOKUSEI_R_LIV', 'T_GAPPEI_R_LIV', 'T_KANSEN_R_LIV', 'T_KENSA_R_LIV', 'T_KENSA_D_LIV', 'T_LIVING_D_LIV', 'T_DONOR_KIDNEY_LIV', 'T_DONOR_LIV', 'T_ISHOKU_KIHON_KIDNEY_LIV', 'T_TRACER_IKO', 'T_ISHOKU_KIHON_LIV' ] # 登録対象テーブル TARGET_TABLES = [ 'T_ISHOKU_KIHON_LIV', ## 値はともかく登録されている 'T_TRACER_IKO', 'T_ISHOKU_KIHON_KIDNEY_LIV', ## 値はともかく登録されている 'T_DONOR_LIV', ## 値はともかく登録されている 'T_DONOR_KIDNEY_LIV', ## 値はともかく登録されている 'T_GAPPEI_R_LIV', # 7/3 実装済 処理はあるが値が入っていない 'T_IJI_MENEKI_YOKUSEI_R_LIV', # 7/3 実装済 処理はあるが値が入っていない 'T_KANSEN_R_LIV', # 7/3 確認済み：一旦実装完了 6/19 'T_LIVING_D_LIV', # 7/3 確認済み：一旦実装完了 6/19 'T_KENSA_D_LIV', # 7/3 実装済：変換表にはないが勝手に移行対象としている 'T_KENSA_R_LIV', # 7/3 実装済：処理はあるが値が入っていない ※まぜるな危険 ] # T_KANSEN_R_LIVの移行対象のデータの組み合わせ kansen_columns = [ ("合併症等:感染症の有無(調査期間中):サイトメガロウイルス抗原血症", "合併症等:感染症の有無(調査期間中):サイトメガロウイルス抗原血症の診断日"), ("合併症等:感染症の有無(調査期間中):サイトメガロウイルス感染症", "合併症等:感染症の有無(調査期間中):サイトメガロウイルス感染症の診断日"), ("合併症等:感染症の有無(調査期間中):その他のウイルス感染症1", "合併症等:感染症の有無(調査期間中):その他のウイルス感染症1の診断日"), ("合併症等:感染症の有無(調査期間中):その他のウイルス感染症2", "合併症等:感染症の有無(調査期間中):その他のウイルス感染症2の診断日"), ("合併症等:感染症の有無(調査期間中):その他のウイルス感染症3", "合併症等:感染症の有無(調査期間中):その他のウイルス感染症3の診断日"), ("合併症等:感染症の有無(調査期間中):細菌性肺炎", "合併症等:感染症の有無(調査期間中):細菌性肺炎の診断日"), ("合併症等:感染症の有無(調査期間中):細菌性尿路感染症", "合併症等:感染症の有無(調査期間中):細菌性尿路感染症の診断日"), ("合併症等:感染症の有無(調査期間中):その他の細菌性感染症1", "合併症等:感染症の有無(調査期間中):その他の細菌性感染症1の診断日"), ("合併症等:感染症の有無(調査期間中):その他の細菌性感染症2", "合併症等:感染症の有無(調査期間中):その他の細菌性感染症2の診断日"), ("合併症等:感染症の有無(調査期間中):その他の細菌性感染症3", "合併症等:感染症の有無(調査期間中):その他の細菌性感染症3の診断日"), ("合併症等:感染症の有無(調査期間中):PC肺炎", "合併症等:感染症の有無(調査期間中):PC肺炎の診断日"), ("合併症等:感染症の有無(調査期間中):その他の真菌性感染症1", "合併症等:感染症の有無(調査期間中):その他の真菌性感染症1の診断日"), ("合併症等:感染症の有無(調査期間中):その他の真菌性感染症2", "合併症等:感染症の有無(調査期間中):その他の真菌性感染症2の診断日"), ("合併症等:感染症の有無(調査期間中):その他の感染症1", "合併症等:感染症の有無(調査期間中):その他の感染症1の診断日"), ("合併症等:感染症の有無(調査期間中):その他の感染症2", "合併症等:感染症の有無(調査期間中):その他の感染症2の診断日"), ("合併症等:感染症の有無(調査期間中):その他の感染症3", "合併症等:感染症の有無(調査期間中):その他の感染症3の診断日"), ] # 臓器コードと移植種別コード ORGAN_CODE_LIVER = "3" ORGAN_CODE_KIDNEY = "4" TRANSPLANT_TYPE_LIVING = "1" # 変換ルールファイル CONVERSION_RULES_FILE = "conversion_rules/t_donor_kidney_liv_rules.json" ########################################################################################## ### 変換処理関数 ########################################################################################## def apply_json_conversion_rules(table_name: str, transplant_row: pd.Series, recipient_rows: List[tuple] = None, donor_rows: List[tuple] = None, conversion_processor: ConversionProcessor = None) -> Dict[str, Any]: """ JSON変換ルールを適用してデータを変換する Args: table_name: 対象テーブル名 transplant_row: 移植時データ recipient_rows: レシピエントフォローアップデータ（降順ソート済み） donor_rows: ドナーフォローアップデータ（降順ソート済み） conversion_processor: 変換プロセッサー Returns: 変換後のデータ辞書 """ if not conversion_processor: logger.warning("変換プロセッサーが設定されていません") return {} try: converted_data = conversion_processor.apply_conversion_rules( table_name, transplant_row, recipient_rows, donor_rows ) logger.debug(f"変換結果 ({table_name}): {len(converted_data)} 項目") return converted_data except Exception as e: logger.error(f"変換処理中にエラーが発生しました: {e}") return {} def update_table_with_converted_data(cursor: pymysql.cursors.DictCursor, connection: pymysql.connections.Connection, table_name: str, primary_key: str, primary_value: Any, converted_data: Dict[str, Any], error_logs: List[Dict]): """ 変換されたデータでテーブルを更新する Args: cursor: データベースカーソル connection: データベース接続 table_name: テーブル名 primary_key: 主キーカラム名 primary_value: 主キー値 converted_data: 変換されたデータ error_logs: エラーログリスト """ if not converted_data: logger.debug(f"更新対象のデータがありません: {table_name}") return try: # SET句を構築 set_clauses = [] values = [] for col, value in converted_data.items(): if isinstance(value, dict): # 複数フラグの場合、各フラグを個別に更新 for flag_col, flag_value in value.items(): set_clauses.append(f"{flag_col} = %s") values.append(flag_value) else: set_clauses.append(f"{col} = %s") values.append(value) if not set_clauses: logger.debug(f"更新対象のカラムがありません: {table_name}") return # UPDATE文を実行 sql = f""" UPDATE {table_name} SET {', '.join(set_clauses)} WHERE {primary_key} = %s """ values.append(primary_value) logger.debug(f"UPDATE SQL: {sql}") logger.debug(f"UPDATE VALUES: {values}") cursor.execute(sql, values) connection.commit() logger.info(f"✅ {table_name} 更新完了: {len(set_clauses)} カラム") except Exception as e: logger.error(f"❌ {table_name} 更新失敗: {e}") error_logs.append({ "テーブル名": table_name, "SQL": sql, "入力値": values, "エラー内容": str(e) }) connection.rollback() ########################################################################################## ### ユーティリティ関数 ########################################################################################## def get_kensa_metadata(csv_column_name: str) -> tuple: """検査項目のメタデータを取得""" metadata_map = { 'AST[追跡時の検査値]': ('AST', 'IU/L', 'int', 40), 'ALT[追跡時の検査値]': ('ALT', 'IU/L', 'int', 40), 'T.Bil[追跡時の検査値]': ('T.Bil', 'mg/dl', 'float', 1.2), 'D.Bil[追跡時の検査値]': ('D.Bil', 'mg/dl', 'float', 0.5), 'ワーファリンの使用[追跡時の検査値]': ('ワーファリン使用', '', 'bool', None), 'INR[追跡時の検査値]': ('INR', '', 'float', 2.0), } if csv_column_name in metadata_map: return metadata_map[csv_column_name] else: return (csv_column_name, '', 'str', None) def evaluate_kensa_code(value: Any, dtype: str, threshold: Optional[float]) -> str: """検査値を評価してコードを返す""" if pd.isna(value) or value == '': return '9' # 未測定 if dtype == 'bool': return '1' if value in ['TRUE', True, '1', 1] else '0' if dtype in ['int', 'float'] and threshold is not None: try: numeric_value = float(value) return '1' if numeric_value > threshold else '0' # 1:異常, 0:正常 except (ValueError, TypeError): return '9' return '0' # デフォルト def get_sisetu_cd(facility_name: str, institution_df: pd.DataFrame) -> str: """施設名から施設コードを取得""" # カラム名の候補リスト name_columns = ['institution_name', '施設名', 'facility_name', 'name'] sisetu_cd_columns = ['SISETU_CD', '施設コード', 'facility_code', 'code'] # 施設名カラムを特定 name_col = None for col in name_columns: if col in institution_df.columns: name_col = col break # 施設コードカラムを特定 code_col = None for col in sisetu_cd_columns: if col in institution_df.columns: code_col = col break # カラムが見つからない場合 if not name_col or not code_col: logger.warning(f"施設マスタに必要なカラムが見つかりません。デフォルト値を返します。") logger.debug(f"利用可能なカラム: {institution_df.columns.tolist()}") return "700000" # 完全一致で検索 match = institution_df[institution_df[name_col] == facility_name] if not match.empty: return match.iloc[0][code_col] # 部分一致で検索 match_like = institution_df[institution_df[name_col].str.contains(facility_name, na=False)] if not match_like.empty: return match_like.iloc[0][code_col] # デフォルト値 return "700000" def render_sql(sql_template: str, values: tuple) -> str: """SQLテンプレートと値を結合して実行可能なSQLを生成（デバッグ用）""" parts = sql_template.split("%s") rendered = "" for i, part in enumerate(parts[:-1]): rendered += part + repr(values[i]) rendered += parts[-1] return rendered def to_nullable_float(val): try: return float(val) except (TypeError, ValueError): return None def convert_pressure_status(val: Optional[str]) -> Optional[int]: mapping = { "良好": 1, "変化なし": 2, "不良": 3, "不明": 4 } return mapping.get(val.strip() if isinstance(val, str) else None, None) def parse_followup_cycle(val: str) -> int: """例: '3ヶ月後' -> 3, '2年後' -> 24""" if "ヶ月" in val: return 0 elif "年" in val: return int(val.replace("年後", "").strip()) return 99999 # 想定外は一番後ろ def to_cycle_str(tsuiseki: str) -> Optional[str]: """ '3ヶ月後' → 99, '1年後' → '01', '2年後' → '02', ..., '50年後' → '50' """ tsuiseki = tsuiseki.strip() if tsuiseki == "3ヶ月後": return "99" elif "年後" in tsuiseki: try: year = int(tsuiseki.replace("年後", "").strip()) return f"{year + 1:02}" # +1 することで3ヶ月後→01, 1年後→02, 2年後→03になる except: return "99" return "99" # 未知の値 ########################################################################################## ### データベース操作関数 ########################################################################################## def reset_tables(cursor: pymysql.cursors.DictCursor) -> None: """全テーブルのデータを削除し、AUTO_INCREMENTをリセット""" cursor.execute("SET FOREIGN_KEY_CHECKS=0;") for table in ORDERED_TABLES: try: cursor.execute(f"DELETE FROM {table};") cursor.execute(f"ALTER TABLE {table} AUTO_INCREMENT = 1;") logger.info(f"テーブル {table} を初期化しました") except Exception as e: logger.error(f"テーブル {table} の初期化に失敗: {e}") cursor.execute("SET FOREIGN_KEY_CHECKS=1;") logger.info("✅ 外部キー制約を考慮したテーブル初期化完了") def execute_insert(cursor: pymysql.cursors.DictCursor, connection: pymysql.connections.Connection, table_name: str, columns: List[str], values: List[Any], error_logs: List[Dict]) -> Optional[int]: """INSERT文を実行し、エラーログを記録""" sql = f""" INSERT INTO {table_name} ({",".join(columns)}) VALUES ({",".join(["%s"] * len(columns))}) """ try: logger.debug(f"実行SQL: {render_sql(sql, values)}") cursor.execute(sql, values) connection.commit() return cursor.lastrowid except Exception as e: logger.error(f"❌ INSERT失敗: {table_name}, エラー: {e}") error_logs.append({ "テーブル名": table_name, "SQL": sql, "入力値": values, "エラー内容": str(e) }) connection.rollback() return None def generate_tracer_id(cursor: pymysql.cursors.DictCursor, transplant_date: str) -> str: """TRACER IDを生成""" year = transplant_date[:4] if transplant_date else "2024" prefix = f"{ORGAN_CODE_KIDNEY}{TRANSPLANT_TYPE_LIVING}{year}" cursor.execute( "SELECT MAX(TRACER_ID) AS max_id FROM T_ISHOKU_KIHON_LIV WHERE TRACER_ID LIKE %s", (f"{prefix}____",) ) result = cursor.fetchone() if result["max_id"]: last_seq = int(result["max_id"][-4:]) next_seq = last_seq + 1 else: next_seq = 1 return f"{prefix}{next_seq:04d}" def insert_ishoku_kihon(cursor: pymysql.cursors.DictCursor, connection: pymysql.connections.Connection, transplant_raw: pd.Series, error_logs: List[Dict]) -> Optional[int]: """T_ISHOKU_KIHON_LIVテーブルへの挿入""" columns = ["ISYOKU_ISYOKUSISETU_CD", "ZOKI_CODE","DEL_FLG", "INS_USER_ID", "INS_PROGRAM_ID"] values = [ transplant_raw["ISYOKU_ISYOKUSISETU_CD"], ORGAN_CODE_KIDNEY, # ZOKI_CODE 0, # DEL_FLG 1, # INS_USER_ID 1 # INS_PROGRAM_ID ] seitai_ishoku_id = execute_insert(cursor, connection, "T_ISHOKU_KIHON_LIV", columns, values, error_logs) if seitai_ishoku_id: # TRACER_ID生成 tracer_id = generate_tracer_id(cursor, transplant_raw.get("移植日", "")) # RECIPIENT_ID = 7桁ゼロ埋め recipient_id = str(seitai_ishoku_id).zfill(7) # UPDATE文：TRACER_ID + RECIPIENT_ID を同時に更新 cursor.execute( """ UPDATE T_ISHOKU_KIHON_LIV SET TRACER_ID = %s, RECIPIENT_ID = %s WHERE SEITAI_ISYOKU_ID = %s """, (tracer_id, recipient_id, seitai_ishoku_id) ) connection.commit() logger.info(f"✅ SEITAI_ISYOKU_ID = {seitai_ishoku_id}, TRACER_ID = {tracer_id}, RECIPIENT_ID = {recipient_id}") return seitai_ishoku_id, tracer_id ## T_KANSEN_R_LIV テーブルへの登録処理 def insert_kansen_r_liv(cursor, connection, recipient_row, seitai_ishoku_id, error_logs): for kansen_col, date_col in kansen_columns: kansen_raw = recipient_row.get(kansen_col) sindan_date_raw = recipient_row.get(date_col) if pd.isna(kansen_raw) or str(kansen_raw).strip() == "": continue # 感染症が空 or NaN → スキップ # --- KANSEN 値変換 --- kansen_map = { "有": 1, "無": 0, "不明": 8 } kansen = kansen_map.get(str(kansen_raw).strip()) if kansen is None: continue # 定義外の値はスキップ # --- 診断日変換（YYYYMMDD形式） --- try: sindan_date = datetime.strptime(sindan_date_raw, "%Y/%m/%d").strftime("%Y%m%d") except Exception: sindan_date = None if not sindan_date: continue # 診断日が変換できない場合スキップ # --- INSERT 処理 --- insert_columns = [ "SEITAI_ISYOKU_ID", "KANSEN", "SINDAN_DATE", "DEL_FLG", "INS_USER_ID", "INS_PROGRAM_ID" ] insert_values = [ seitai_ishoku_id, kansen, sindan_date, "0", "1", "1" ] sql = f""" INSERT INTO T_KANSEN_R_LIV ({','.join(insert_columns)}) VALUES ({','.join(['%s'] * len(insert_columns))}) """ try: cursor.execute(sql, insert_values) connection.commit() except Exception as e: logger.error(f"❌ INSERT失敗: T_KANSEN_R_LIV, エラー: {e}") print(f"❌ INSERT失敗: T_KANSEN_R_LIV, エラー: {e}") error_logs.append({ "テーブル名": "T_KANSEN_R_LIV", "SQL": sql, "入力値": insert_values, "エラー内容": str(e) }) connection.rollback() def insert_nyuryokujokyo(cursor: pymysql.cursors.DictCursor, connection: pymysql.connections.Connection, seitai_ishoku_id: int, donor_a_id: int) -> None: """T_NYURYOKUJOKYO_LIVテーブルへの挿入""" nyuryoku_rows = [ { "KANJA_KBN": "0", # 1:recipient "KANJA_ID": seitai_ishoku_id, "KIROKU_TIMING": "0", # 0:新規 "NYURYOKUJOKYO": "0", # 0:未入力 }, { "KANJA_KBN": "1", # 1:ドナー "KANJA_ID": donor_a_id, "KIROKU_TIMING": "0", "NYURYOKUJOKYO": "0", } ] for nyuryoku in nyuryoku_rows: sql = """ INSERT INTO T_NYURYOKUJOKYO_LIV ( SEITAI_ISYOKU_ID, KANJA_KBN, KANJA_ID, KIROKU_TIMING, NYURYOKUJOKYO, INS_USER_ID, INS_PROGRAM_ID, INS_DATE, UPD_USER_ID, UPD_PROGRAM_ID, UPD_DATE ) VALUES (%s, %s, %s, %s, %s, %s, %s, NOW(), %s, %s, NOW()) """ values = ( seitai_ishoku_id, nyuryoku["KANJA_KBN"], nyuryoku["KANJA_ID"], nyuryoku["KIROKU_TIMING"], nyuryoku["NYURYOKUJOKYO"], "1", "1", "1", "1" ) cursor.execute(sql, values) connection.commit() ########################################################################################## ### メイン処理 ########################################################################################## def load_data(): """データファイルの読み込み""" try: # CSVファイル読み込み logger.info("施設データを読み込み中...") institution_df = pd.read_csv("/csv/liver/shisetsu.csv", sep="\t", dtype=str).fillna("") # カラム名を確認 logger.info(f"施設データのカラム: {institution_df.columns.tolist()}") logger.info("ドナーフォローアップデータを読み込み中...") donor_df = pd.read_csv("/csv/kan/d-followups.csv", encoding="cp932", dtype=str).fillna("") logger.info(f"ドナーデータ件数: {len(donor_df)}") logger.info("レシピエントフォローアップデータを読み込み中...") recipient_df = pd.read_csv("/csv/kan/r-followups.csv", encoding="cp932", dtype=str).fillna("") logger.info(f"レシピエントデータ件数: {len(recipient_df)}") logger.info("移植データを読み込み中...") transplants_df = pd.read_csv("/csv/kan/transplants.csv", encoding="cp932", dtype=str).fillna("") ## デバック用 # YearNoが2009001以上のデータを抽出し、最初の100件を取得 transplants_df["YearNo_numeric"] = pd.to_numeric(transplants_df["YearNo"], errors="coerce") transplants_df = transplants_df[transplants_df["YearNo_numeric"] >= 20090001] transplants_df = transplants_df.sort_values("YearNo_numeric") transplants_df = transplants_df.head(100) logger.info(f"移植データ件数: {len(transplants_df)}") # 施設名カラムの存在確認 if '施設名' in transplants_df.columns: # 施設コード変換 transplants_df['ISYOKU_ISYOKUSISETU_CD'] = transplants_df['施設名'].apply( lambda x: get_sisetu_cd(x, institution_df) ) else: logger.warning("'施設名'カラムが見つかりません。デフォルト施設コードを使用します。") transplants_df['ISYOKU_ISYOKUSISETU_CD'] = "700000" # Excelファイル読み込み logger.info("マッピングデータを読み込み中...") mapping_dict = pd.read_excel("/csv/kan/移行データ対応表.xlsx", sheet_name=None, header=1) if "腎臓　生体" in mapping_dict: df_mapping = mapping_dict["腎臓　生体"] else: logger.warning("'腎臓　生体'シートが見つかりません。") available_sheets = list(mapping_dict.keys()) logger.info(f"利用可能なシート: {available_sheets}") # デフォルトで最初のシートを使用するか、空のDataFrameを作成 df_mapping = pd.DataFrame() if not mapping_dict else list(mapping_dict.values())[0] return institution_df, donor_df, recipient_df, transplants_df, df_mapping except FileNotFoundError as e: logger.error(f"ファイルが見つかりません: {e}") raise except Exception as e: logger.error(f"データ読み込み中にエラーが発生しました: {e}") raise def process_transplant_data(cursor: pymysql.cursors.DictCursor, connection: pymysql.connections.Connection, transplant_raw: pd.Series, df_mapping: pd.DataFrame, donor_df: pd.DataFrame, recipient_df: pd.DataFrame, institution_df: pd.DataFrame, error_logs: List[Dict]): """1件の移植データを処理""" # T_ISHOKU_KIHON_LIVへの挿入 seitai_ishoku_id, tracer_id = insert_ishoku_kihon(cursor, connection, transplant_raw, error_logs) if not seitai_ishoku_id: return # YearNoを取得（recipient_rowsとdonor_rowsの取得に使用） year_no = transplant_raw.get('YearNo') # 関連するレシピエントとドナーのデータを取得 recipient_rows = [] donor_rows = [] if year_no: # レシピエントデータの取得 if 'レシピエント追跡調査:YearNo' in recipient_df.columns: logger.info(f" recipient {year_no}") recipient_data = recipient_df[recipient_df['レシピエント追跡調査:YearNo'] == year_no] if 'レシピエント追跡調査:追跡調査の種類' in recipient_data.columns: recipient_data["sort_key"] = recipient_data["レシピエント追跡調査:追跡調査の種類"].apply(parse_followup_cycle) recipient_data = recipient_data.sort_values("sort_key") recipient_rows = list(recipient_data.iterrows()) # ドナーデータの取得 if 'ドナー追跡調査:YearNo' in donor_df.columns: logger.info(f" donor {year_no}") donor_data = donor_df[donor_df['ドナー追跡調査:YearNo'] == year_no] if 'ドナー追跡調査:追跡調査の種類' in donor_data.columns: donor_data["sort_key"] = donor_data["ドナー追跡調査:追跡調査の種類"].apply(parse_followup_cycle) donor_data = donor_data.sort_values("sort_key") donor_rows = list(donor_data.iterrows()) # 各テーブルへの処理 donor_a_id = None for target_table in TARGET_TABLES: filtered_mapping = df_mapping[df_mapping["テーブル物理名"] == target_table] if target_table == "T_ISHOKU_KIHON_LIV": continue # 既に処理済み elif target_table == 'T_TRACER_IKO': columns = ["TRACER_ID", "ZOKI_CODE", "SEITAI_ISYOKU_ID", "ISHOKU_TOROKU_ID"] values = [tracer_id,4,seitai_ishoku_id, year_no] execute_insert(cursor, connection, target_table, columns, values, error_logs) if target_table == 'T_ISHOKU_KIHON_KIDNEY_LIV': columns = ["SEITAI_ISYOKU_ID", "INS_USER_ID", "INS_PROGRAM_ID"] values = [seitai_ishoku_id, 1, 1] execute_insert(cursor, connection, target_table, columns, values, error_logs) # TODO: update_table関数が未定義です # TODO: 以下の変数が未定義です: column_series, row, ishoku_toroku_id, recipient_rows, donor_rows # update_table( # cursor=cursor, # connection=connection, # table_name=target_table, # column_series=column_series, # 未定義 # data_row=row, # 未定義 # id_columns="SEITAI_ISYOKU_ID", # id_values=seitai_ishoku_id, # ishoku_toroku_id=ishoku_toroku_id, # 未定義 # recipient_rows=recipient_rows, # 未定義 # donor_rows=donor_rows # 未定義 # ) elif target_table == 'T_DONOR_LIV': # DONOR_IDの生成 cursor.execute("SELECT MAX(DONOR_ID) AS max_id FROM T_DONOR_LIV") result = cursor.fetchone() next_donor_id = int(result['max_id']) + 1 if result['max_id'] else 1 donor_id_str = str(next_donor_id).zfill(7) columns = ["DONOR_ID", "SEITAI_ISYOKU_ID", "INS_USER_ID", "INS_PROGRAM_ID"] values = [donor_id_str, seitai_ishoku_id, 1, 1] donor_a_id = execute_insert(cursor, connection, target_table, columns, values, error_logs) # TODO: update_table関数の呼び出し（未定義） elif target_table == 'T_DONOR_KIDNEY_LIV': # TODO: このテーブル名はマッピングに存在しない可能性があります # T_DONOR_LIVER_LIVと混同している可能性があります if donor_a_id: columns = ["SEITAI_ISYOKU_ID", "DONOR_A_ID", "INS_USER_ID", "INS_PROGRAM_ID"] values = [seitai_ishoku_id, donor_a_id, 1, 1] execute_insert(cursor, connection, target_table, columns, values, error_logs) # TODO: update_table関数の呼び出し（未定義） elif target_table == 'T_DONOR_LIVER_LIV': # 元のコードにあったT_DONOR_LIVER_LIVの処理 if donor_a_id: columns = ["SEITAI_ISYOKU_ID", "DONOR_A_ID", "INS_USER_ID", "INS_PROGRAM_ID"] values = [seitai_ishoku_id, donor_a_id, 1, 1] execute_insert(cursor, connection, target_table, columns, values, error_logs) # TODO: update_table関数の呼び出し（未定義） # T_GAPPEI_R_LIVテーブルへの登録処理（合併症データ） elif target_table == 'T_GAPPEI_R_LIV': logger.info("📥 T_GAPPEI_R_LIVへ合併症データを登録中...") # GAPPEI_Lの判定用マッピング gappei_l_mapping = { "感染症": "01", "悪性腫瘍": "02", "心疾患": "03", "肝障害": "04", "消化性潰瘍": "05", "脳血管障害": "06", "拒絶反応": "07", "再発/de novo": "08" } # レシピエントフォローアップデータを処理 for index__, recipient_row in recipient_rows: # その他の合併症データの処理（1〜3） for i in range(1, 4): # 1, 2, 3 try: # CSVカラム名 gappei_col = f"合併症等:その他の合併症:その他の合併症{i}" date_col = f"合併症等:その他の合併症:その他の合併症{i}の診断日" # 合併症の内容を取得 gappei_content = recipient_row.get(gappei_col) # 値が存在しない場合はスキップ if pd.isna(gappei_content) or str(gappei_content).strip() == "": continue gappei_str = str(gappei_content).strip() # GAPPEI_L（小分類）の判定 gappei_l = "99" # デフォルトはその他 for keyword, code in gappei_l_mapping.items(): if keyword in gappei_str: gappei_l = code break # 最初にマッチしたものを採用 # 診断日の処理 nyuin_date_raw = recipient_row.get(date_col) nyuin_date = "" if pd.notna(nyuin_date_raw) and str(nyuin_date_raw).strip(): try: # 日付フォーマットの変換を試みる # 想定フォーマット: YYYY/MM/DD, YYYY-MM-DD, YYYY年MM月DD日など date_str = str(nyuin_date_raw).strip() # 複数のフォーマットを試す date_formats = [ "%Y/%m/%d", "%Y-%m-%d", "%Y年%m月%d日", "%Y.%m.%d" ] for fmt in date_formats: try: parsed_date = datetime.strptime(date_str, fmt) nyuin_date = parsed_date.strftime("%Y%m%d") break except ValueError: continue if not nyuin_date: logger.warning(f"日付変換失敗: {date_str}") except Exception as e: logger.warning(f"診断日の処理でエラー: {e}") # INSERT処理 insert_columns = [ "SEITAI_ISYOKU_ID", "GAPPEI", "GAPPEI_L", "NYUIN_DATE", "CMNT", "DEL_FLG", "INS_USER_ID", "INS_PROGRAM_ID" ] insert_values = [ seitai_ishoku_id, "93", gappei_l, nyuin_date, gappei_str[:120], # CMNTは最大120文字 "0", 1, 1 ] sql = f""" INSERT INTO T_GAPPEI_R_LIV ({','.join(insert_columns)}) VALUES ({','.join(['%s'] * len(insert_columns))}) """ cursor.execute(sql, insert_values) connection.commit() logger.debug(f"✅ 合併症{i}を登録: GAPPEI=93, GAPPEI_L={gappei_l}, 内容={gappei_str[:50]}...") except Exception as e: logger.error(f"❌ INSERT失敗: T_GAPPEI_R_LIV (合併症{i}), エラー: {e}") error_logs.append({ "テーブル名": "T_GAPPEI_R_LIV", "エラー内容": str(e), "SQL": sql, "入力値": insert_values, "合併症番号": i, "YearNo": year_no }) connection.rollback() logger.info(f"✅ T_GAPPEI_R_LIV登録完了: SEITAI_ISYOKU_ID={seitai_ishoku_id}") # T_IJI_MENEKI_YOKUSEI_R_LIVテーブルへの登録処理（免疫抑制薬データ） elif target_table == 'T_IJI_MENEKI_YOKUSEI_R_LIV': logger.info("📥 T_IJI_MENEKI_YOKUSEI_R_LIVへ免疫抑制薬データを登録中...") # レシピエントフォローアップデータを処理 for index__, recipient_row in recipient_rows: # 追跡調査の種類からCYCLEを決定 tsuiseki_shurui = recipient_row.get("レシピエント追跡調査:追跡調査の種類", "") cycle_str = to_cycle_str(tsuiseki_shurui) # スキップ条件：CYCLEが取得できない場合 if not cycle_str or cycle_str == "99": logger.warning(f"⚠️ CYCLEが不正なためスキップ: {tsuiseki_shurui}") continue logger.debug(f"処理中: YearNo={year_no}, 追跡調査種類={tsuiseki_shurui}, CYCLE={cycle_str}") try: # 初期値設定（すべて0で初期化） cs = 0 csa = 0 tac = 0 tac_type = None rap = 0 az = 0 cp = 0 mmf = 0 mz = 0 evl = 0 # SQLログにはあるが、CSVルールに記載なし（0固定） meneki_etc = 0 meneki_etc_cmnt = None # ①ステロイド（CS） steroid_val = recipient_row.get("免疫抑制剤:ステロイド", "") if pd.notna(steroid_val) and str(steroid_val).strip() == "有": cs = 1 # ②カルシニュリン・インヒビター（CSA, TAC） calcineurin_val = recipient_row.get("免疫抑制剤:カルシニュリン・インヒビター", "") if pd.notna(calcineurin_val): cal_str = str(calcineurin_val).strip() # 不明・未使用はスキップ if cal_str not in ["不明", "未使用", ""]: if "シクロスポリン" in cal_str or "CyA" in cal_str: csa = 1 if "タクロリムス" in cal_str or "FK506" in cal_str: tac = 1 # TAC_TYPEの設定（FK506→1, タクロリムス-ER→2, 不明→3） if "FK506" in cal_str: tac_type = "1" elif "タクロリムス-ER" in cal_str: tac_type = "2" else: tac_type = "3" # その他のタクロリムスは不明扱い # ③mTOR阻害剤（RAP） mtor_val = recipient_row.get("免疫抑制剤:mTOR阻害剤", "") if pd.notna(mtor_val): mtor_str = str(mtor_val).strip() # 不明・未使用はスキップ、それ以外の値があれば1 if mtor_str not in ["不明", "未使用", ""] and mtor_str: rap = 1 # ④核酸合成阻害剤（AZ, CP, MMF, MZ） nucleic_val = recipient_row.get("免疫抑制剤:核酸合成阻害剤", "") if pd.notna(nucleic_val): nuc_str = str(nucleic_val).strip() # 不明・未使用はスキップ if nuc_str not in ["不明", "未使用", ""]: # 文字列に含まれる薬剤をチェック if "AZP" in nuc_str or "アザチオプリン" in nuc_str: az = 1 if "CP" in nuc_str or "シクロホスファミド" in nuc_str: cp = 1 if "MMF" in nuc_str or "ミコフェノール酸モフェチル" in nuc_str: mmf = 1 if "MZR" in nuc_str or "ミゾリビン" in nuc_str: mz = 1 # ⑤その他の使用（MENEKI_ETC） other_use_val = recipient_row.get("免疫抑制剤:その他の使用", "") if pd.notna(other_use_val): other_str = str(other_use_val).strip() # 不明・未使用はスキップ、それ以外の値があれば1 if other_str not in ["不明", "未使用", ""] and other_str: meneki_etc = 1 # ⑥薬剤名（MENEKI_ETC_CMNT） drug_name_val = recipient_row.get("免疫抑制剤:薬剤名", "") if pd.notna(drug_name_val) and str(drug_name_val).strip(): meneki_etc_cmnt = str(drug_name_val).strip()[:45] # 最大45文字 # DSA関連（SQLログにはあるが、CSVルールに記載なし - NULL設定） dsa = None dsa_kensa = None dsa_kogen1 = None dsa_mfi1 = None dsa_kogen2 = None dsa_mfi2 = None dsa_kogen3 = None dsa_mfi3 = None trough_cni = None trough_mtor = None # INSERT処理 insert_columns = [ "SEITAI_ISYOKU_ID", "CYCLE", "CSA", "TAC", "TAC_TYPE", "CS", "MMF", "EVL", "AZ", "MZ", "RAP", "CP", "MENEKI_ETC", "MENEKI_ETC_CMNT", "DSA", "DSA_KENSA", "DSA_KOGEN1", "DSA_MFI1", "DSA_KOGEN2", "DSA_MFI2", "DSA_KOGEN3", "DSA_MFI3", "TROUGH_CNI", "TROUGH_MTOR", "INS_USER_ID", "INS_PROGRAM_ID" ] insert_values = [ seitai_ishoku_id, cycle_str, str(csa), str(tac), tac_type, str(cs), str(mmf), str(evl), str(az), str(mz), str(rap), str(cp), str(meneki_etc), meneki_etc_cmnt, dsa, dsa_kensa, dsa_kogen1, dsa_mfi1, dsa_kogen2, dsa_mfi2, dsa_kogen3, dsa_mfi3, trough_cni, trough_mtor, 1, 1 ] sql = f""" INSERT INTO T_IJI_MENEKI_YOKUSEI_R_LIV ({','.join(insert_columns)}) VALUES ({','.join(['%s'] * len(insert_columns))}) """ cursor.execute(sql, insert_values) connection.commit() # デバッグログ logger.debug(f"✅ 免疫抑制薬データを登録:") logger.debug(f" CS={cs}, CSA={csa}, TAC={tac}, TAC_TYPE={tac_type}") logger.debug(f" RAP={rap}, AZ={az}, CP={cp}, MMF={mmf}, MZ={mz}") logger.debug(f" MENEKI_ETC={meneki_etc}, MENEKI_ETC_CMNT={meneki_etc_cmnt}") except Exception as e: logger.error(f"❌ INSERT失敗: T_IJI_MENEKI_YOKUSEI_R_LIV, エラー: {e}") error_logs.append({ "テーブル名": "T_IJI_MENEKI_YOKUSEI_R_LIV", "エラー内容": str(e), "SQL": sql, "入力値": insert_values, "CYCLE": cycle_str, "YearNo": year_no }) connection.rollback() logger.info(f"✅ T_IJI_MENEKI_YOKUSEI_R_LIV登録完了: SEITAI_ISYOKU_ID={seitai_ishoku_id}") elif target_table == 'T_KANSEN_R_LIV': for _, recipient_row in recipient_rows: insert_kansen_r_liv(cursor, connection, recipient_row, seitai_ishoku_id, error_logs) elif target_table == 'T_LIVING_D_LIV': for _, donor_row in donor_rows: row = donor_row # pandas.Series # --- INPUT_DATE の変換 --- input_date_raw = row.get("ドナー追跡調査:調査日") try: input_date = datetime.strptime(input_date_raw, "%Y/%m/%d").strftime("%Y%m%d") except Exception: input_date = "19000101" # 万が一エラー時のデフォルト（異常値として識別可能） donor_cycle = row.get("ドナー追跡調査:追跡調査の種類") cycle_str = to_cycle_str(donor_cycle) if not cycle_str or cycle_str.strip() == "99": continue # --- その他の値 --- height = to_nullable_float(row.get("ドナー所見:身長")) weight = to_nullable_float(row.get("ドナー所見:体重")) upper = to_nullable_float(row.get("ドナー所見:収縮期")) lower = to_nullable_float(row.get("ドナー所見:拡張期")) shakaifukki_seishin = convert_pressure_status(row.get("ドナー所見:社会復帰状況:精神的")) shakaifukki_shintai = convert_pressure_status(row.get("ドナー所見:社会復帰状況:身体的")) sisetu_cd = get_sisetu_cd(transplant_raw.get('ドナー予後:生存:施設名', ""), institution_df) insert_columns = [ "DONOR_A_ID", "INPUT_DATE", "SISETU_CD", "HEIGHT", "WEIGHT", "KETSUKATSU_UPPER", "KETSUKATSU_LOWER", "SHAKAIFUKKI_JOKYO_SESHIN", "SHAKAIFUKKI_JOKYO_SHINTAI", "CYCLE", "DEL_FLG", "INS_USER_ID", "INS_PROGRAM_ID" ] insert_values = [ donor_a_id, input_date, sisetu_cd, height, weight, upper, lower, shakaifukki_seishin, shakaifukki_shintai, cycle_str, "0", "1", "1" ] sql = f""" INSERT INTO T_LIVING_D_LIV ({','.join(insert_columns)}) VALUES ({','.join(['%s'] * len(insert_columns))}) """ try: cursor.execute(sql, insert_values) connection.commit() except Exception as e: logger.error(f"❌ INSERT失敗: T_LIVING_D_LIV, エラー: {e}") error_logs.append({ "テーブル名": "T_LIVING_D_LIV", "SQL": sql, "入力値": insert_values, "エラー内容": str(e) }) connection.rollback() # 2025.07.02　実装 T_KENSA_D_LIVテーブルへの登録処理を追加する部分 elif target_table == 'T_KENSA_D_LIV': logger.info("📥 T_KENSA_D_LIVへドナー検査データを登録中...") if not donor_a_id: logger.warning("⚠️ DONOR_A_IDが存在しないため、T_KENSA_D_LIVへの登録をスキップします") continue # ドナー検査項目のマッピング定義 # (CSVカラム名, KENSA_NAME, KENSA_UNIT) donor_kensa_mappings = [ # 血液検査 ("ドナー所見:血液検査:WBC", "WBC", "/μL"), ("ドナー所見:血液検査:Hb", "Hb", "g/dL"), ("ドナー所見:血液検査:Ht", "Ht", "%"), ("ドナー所見:血液検査:Bun", "BUN", "mg/dL"), ("ドナー所見:血液検査:シスタチンC", "シスタチン C", "mg/L"), ("ドナー所見:血液検査:TC", "TC", "mg/dL"), ("ドナー所見:血液検査:LDL-C", "LDL-C", "mg/dL"), ("ドナー所見:血液検査:HbA1c", "HbA1c(NGSP)", "%"), # 腎機能 ("ドナー所見:腎機能:血清Cr", "血清Cr", "mg/dL"), # 尿検査 ("ドナー所見:尿検査:尿中蛋白", "尿中蛋白", "mg/dL"), ("ドナー所見:尿検査:尿中Cr定量", "尿中Cr（定量）", "mg/dL"), ("ドナー所見:尿検査:Na", "Na", "mEq/L"), ("ドナー所見:尿検査:K", "K", "mEq/L"), ("ドナー所見:尿検査:Cl", "Cl", "mEq/L"), ("ドナー所見:尿検査:Cr", "尿Cr", "mg/dL"), ("ドナー所見:尿検査:UUN", "尿中尿素窒素", "mg/dL"), ("ドナー所見:尿検査:尿糖", "尿糖", ""), ("ドナー所見:尿検査:尿ケトン", "尿ケトン", ""), ("ドナー所見:尿検査:尿沈渣", "尿沈渣", ""), # 腎機能追加 ("ドナー所見:腎機能:24hCr・クリアランス", "24hCr・クリアランス", "mL/min"), ("ドナー所見:腎機能:イヌリン・クリアランス", "イヌリン・クリアランス", "mL/min"), # 動脈硬化検査 ("ドナー所見:動脈硬化:PWV(右・左)", "PWV(右・左)", "cm/sec"), ("ドナー所見:動脈硬化:PWV", "PWV", "cm/sec"), ("ドナー所見:動脈硬化:ABI(右・左)", "ABI(右・左)", ""), ("ドナー所見:動脈硬化:ABI", "ABI", ""), # その他固定項目（SQLログから確認された項目） ("", "eGFR", "mL/min/1.73㎡"), # 計算値のため空 ("", "透析導入", ""), # フラグのため空 ("", "尿中アルブミン", "mg/dL"), ("", "尿中β2 ミクログロブリン", "μg/L"), ] # 癌検診項目の定義（有無のチェック項目） gan_kenshin_mappings = [ ("ドナー所見:癌検診:胃内視鏡", "胃内視鏡", "ドナー所見:癌検診:胃内視鏡 有の場合"), ("ドナー所見:癌検診:乳房撮影", "乳房撮影", "ドナー所見:癌検診:乳房撮影 有の場合"), ("ドナー所見:癌検診:便潜血", "便潜血", "ドナー所見:癌検診:便潜血 有の場合"), ("ドナー所見:癌検診:子宮癌検査", "子宮癌検査", "ドナー所見:癌検診:子宮癌検査 有の場合"), ("ドナー所見:癌検診:腹部超音波", "腹部超音波", "ドナー所見:癌検診:腹部超音波 有の場合"), ] # 各ドナーフォローアップデータを処理 for index__, donor_row in donor_rows: # 追跡調査の種類からCYCLEを決定 tsuiseki_shurui = donor_row.get("ドナー追跡調査:追跡調査の種類", "") cycle_str = to_cycle_str(tsuiseki_shurui) logger.debug(f"処理中: YearNo={year_no}, 追跡調査種類={tsuiseki_shurui}, CYCLE={cycle_str}") # 一般検査項目の処理 for csv_col, kensa_name, kensa_unit in donor_kensa_mappings: try: # CSVカラムが空の場合は固定値を設定 if csv_col == "": # eGFR, 透析導入などの特殊処理 if kensa_name == "eGFR": # eGFRは計算値のため、血清Crから計算するか、デフォルト値を設定 kensa_value = "" elif kensa_name == "透析導入": # 透析導入フラグ kensa_value = "" else: # その他の固定項目 kensa_value = "" else: # CSVから値を取得 kensa_value = donor_row.get(csv_col) # 値が存在しない場合はスキップ if pd.isna(kensa_value) or kensa_value == "": continue # KENSA_VALUE_CODEの設定（通常はNULL） kensa_value_code = None # 検査日の設定（通常はNULL） kensa_date = None # 検査フラグの設定（通常はNULL） kensa_flg = None # INSERT処理 insert_columns = [ "DONOR_A_ID", "KENSA_NAME", "CYCLE", "KENSA_VALUE", "KENSA_UNIT", "KENSA_VALUE_CODE", "KENSA_DATE", "KENSA_FLG", "INS_USER_ID", "INS_PROGRAM_ID" ] insert_values = [ donor_a_id, kensa_name, cycle_str, str(kensa_value), kensa_unit, kensa_value_code, kensa_date, kensa_flg, 1, 1 ] sql = f""" INSERT INTO T_KENSA_D_LIV ({','.join(insert_columns)}) VALUES ({','.join(['%s'] * len(insert_columns))}) """ cursor.execute(sql, insert_values) connection.commit() logger.debug(f"✅ {kensa_name}を登録: 値={kensa_value}") except Exception as e: logger.error(f"❌ INSERT失敗: T_KENSA_D_LIV ({kensa_name}), エラー: {e}") error_logs.append({ "テーブル名": "T_KENSA_D_LIV", "検査項目": kensa_name, "SQL": sql, "入力値": insert_values, "エラー内容": str(e) }) connection.rollback() # 癌検診項目の処理（有無チェック） for csv_col, kensa_name, comment_col in gan_kenshin_mappings: try: # 検査結果（異常無/異常有/未検） kensa_result = donor_row.get(csv_col) # 有の場合のコメント kensa_comment = donor_row.get(comment_col, "") if pd.isna(kensa_result) or kensa_result == "": continue # KENSA_VALUE_CODEの設定 # 癌検診の場合は結果をコード化 value_code_map = { "異常無": "1", "異常有": "2", "未検": "3" } kensa_value_code = value_code_map.get(str(kensa_result).strip(), None) if not kensa_value_code: continue # INSERT処理 insert_columns = [ "DONOR_A_ID", "KENSA_NAME", "CYCLE", "KENSA_VALUE", "KENSA_UNIT", "KENSA_VALUE_CODE", "KENSA_DATE", "KENSA_FLG", "CMNT", "INS_USER_ID", "INS_PROGRAM_ID" ] insert_values = [ donor_a_id, kensa_name, cycle_str, "", "", kensa_value_code, None, None, kensa_comment if kensa_comment else None, 1, 1 ] sql = f""" INSERT INTO T_KENSA_D_LIV ({','.join(insert_columns)}) VALUES ({','.join(['%s'] * len(insert_columns))}) """ cursor.execute(sql, insert_values) connection.commit() logger.debug(f"✅ 癌検診 {kensa_name}を登録: 結果={kensa_result}") except Exception as e: logger.error(f"❌ INSERT失敗: T_KENSA_D_LIV (癌検診 {kensa_name}), エラー: {e}") error_logs.append({ "テーブル名": "T_KENSA_D_LIV", "検査項目": f"癌検診_{kensa_name}", "SQL": sql, "入力値": insert_values, "エラー内容": str(e) }) connection.rollback() logger.info(f"✅ T_KENSA_D_LIV登録完了: DONOR_A_ID={donor_a_id}") # T_KENSA_R_LIVテーブルへの登録処理（レシピエント検査データ） elif target_table == 'T_KENSA_R_LIV': logger.info("📥 T_KENSA_R_LIVへレシピエント検査データを登録中...") # レシピエントフォローアップデータを処理 for index__, recipient_row in recipient_rows: # 追跡調査の種類からCYCLEを決定 tsuiseki_shurui = recipient_row.get("レシピエント追跡調査:追跡調査の種類", "") cycle_str = to_cycle_str(tsuiseki_shurui) logger.debug(f"処理中: YearNo={year_no}, 追跡調査種類={tsuiseki_shurui}, CYCLE={cycle_str}") # 1. 血清Crの処理 serum_cr = recipient_row.get("移植腎予後:血清Cr") if pd.notna(serum_cr) and str(serum_cr).strip() != "": try: # 血清Crは数値としてそのまま格納 insert_columns = [ "SEITAI_ISYOKU_ID", "KENSA_NAME", "CYCLE", "KENSA_VALUE", "KENSA_UNIT", "KENSA_VALUE_CODE", "KENSA_DATE", "CMNT", "INS_USER_ID", "INS_PROGRAM_ID" ] insert_values = [ seitai_ishoku_id, "血清Cr", cycle_str, str(serum_cr), "mg/dL", None, # KENSA_VALUE_CODEはNULL None, None, 1, 1 ] sql = f""" INSERT INTO T_KENSA_R_LIV ({','.join(insert_columns)}) VALUES ({','.join(['%s'] * len(insert_columns))}) """ cursor.execute(sql, insert_values) connection.commit() logger.debug(f"✅ 血清Crを登録: 値={serum_cr}") except Exception as e: logger.error(f"❌ INSERT失敗: T_KENSA_R_LIV (血清Cr), エラー: {e}") error_logs.append({ "テーブル名": "T_KENSA_R_LIV", "検査項目": "血清Cr", "SQL": sql, "入力値": insert_values, "エラー内容": str(e) }) connection.rollback() # 2. 尿中蛋白の処理 urine_protein = recipient_row.get("移植腎予後:尿中蛋白") if pd.notna(urine_protein) and str(urine_protein).strip() != "": try: # 尿中蛋白の値をコード化 protein_value = str(urine_protein).strip() # KENSA_VALUE_CODEのマッピング if protein_value == "-": kensa_value_code = "1" # 陰性 elif protein_value == "+": kensa_value_code = "2" # 陽性 elif protein_value in ["不明", "未検", "未測定"]: kensa_value_code = "3" # 不明 else: # その他の値の場合（±、++、+++など） kensa_value_code = "2" # 陽性として扱う # 尿中蛋白の登録 insert_columns = [ "SEITAI_ISYOKU_ID", "KENSA_NAME", "CYCLE", "KENSA_VALUE", "KENSA_UNIT", "KENSA_VALUE_CODE", "KENSA_DATE", "CMNT", "INS_USER_ID", "INS_PROGRAM_ID" ] insert_values = [ seitai_ishoku_id, "尿中蛋白", cycle_str, "1", "mg/dL", kensa_value_code, None, None, 1, 1 ] sql = f""" INSERT INTO T_KENSA_R_LIV ({','.join(insert_columns)}) VALUES ({','.join(['%s'] * len(insert_columns))}) """ cursor.execute(sql, insert_values) connection.commit() logger.debug(f"✅ 尿中蛋白を登録: 値={protein_value}, コード={kensa_value_code}") # 尿中アルブミンも同時に登録（SQLログから確認された仕様） insert_values_albumin = [ seitai_ishoku_id, "尿中アルブミン", cycle_str, "2", "mg/dL", kensa_value_code, # 同じKENSA_VALUE_CODE None, None, 1, 1 ] sql_albumin = f""" INSERT INTO T_KENSA_R_LIV ({','.join(insert_columns)}) VALUES ({','.join(['%s'] * len(insert_columns))}) """ cursor.execute(sql_albumin, insert_values_albumin) connection.commit() logger.debug(f"✅ 尿中アルブミンを登録: コード={kensa_value_code}") except Exception as e: logger.error(f"❌ INSERT失敗: T_KENSA_R_LIV (尿中蛋白/アルブミン), エラー: {e}") error_logs.append({ "テーブル名": "T_KENSA_R_LIV", "検査項目": "尿中蛋白/アルブミン", "SQL": sql, "入力値": insert_values, "エラー内容": str(e) }) connection.rollback() logger.info(f"✅ T_KENSA_R_LIV登録完了: SEITAI_ISYOKU_ID={seitai_ishoku_id}") # T_NYURYOKUJOKYO_LIVへの挿入 if donor_a_id: insert_nyuryokujokyo(cursor, connection, seitai_ishoku_id, donor_a_id) def debug_csv_structure(file_path: str, encoding: str = "utf-8", sep: str = ",", nrows: int = 5): """CSVファイルの構造をデバッグ出力""" try: df = pd.read_csv(file_path, encoding=encoding, sep=sep, nrows=nrows) logger.info(f"\n=== {file_path} の構造 ===") logger.info(f"カラム: {df.columns.tolist()}") logger.info(f"データ型: \n{df.dtypes}") logger.info(f"先頭{nrows}行のデータ:\n{df.head()}") logger.info("=" * 50) except Exception as e: logger.error(f"{file_path} の読み込みに失敗: {e}") def output_error_logs(error_logs: List[Dict]) -> None: """エラーログをファイルとコンソールに出力""" if not error_logs: logger.info("✅ エラーなく処理が完了しました") return # エラーサマリーの出力 logger.error(f"❌ 処理中にエラーが発生しました: 合計 {len(error_logs)} 件") # テーブル別エラー集計 table_error_counts = {} for error in error_logs: table_name = error.get("テーブル名", "不明") table_error_counts[table_name] = table_error_counts.get(table_name, 0) + 1 logger.error("【テーブル別エラー件数】") for table, count in sorted(table_error_counts.items()): logger.error(f" - {table}: {count}件") # エラー詳細をログファイルに出力 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") error_log_filename = f"./logs/migration_errors_{timestamp}.log" try: # ログディレクトリの作成（存在しない場合） os.makedirs("./logs", exist_ok=True) with open(error_log_filename, 'w', encoding='utf-8') as f: f.write(f"=== 移行処理エラーログ ===\n") f.write(f"実行日時: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n") f.write(f"エラー総数: {len(error_logs)}件\n") f.write("=" * 50 + "\n\n") for i, error in enumerate(error_logs, 1): f.write(f"【エラー {i}/{len(error_logs)}】\n") f.write(f"テーブル名: {error.get('テーブル名', '不明')}\n") # 検査項目名がある場合（T_KENSA系テーブル） if "検査項目" in error: f.write(f"検査項目: {error.get('検査項目')}\n") f.write(f"エラー内容: {error.get('エラー内容', '不明')}\n") f.write(f"SQL: {error.get('SQL', '不明')}\n") f.write(f"入力値: {error.get('入力値', '不明')}\n") f.write("-" * 50 + "\n\n") logger.info(f"📄 エラー詳細をファイルに出力しました: {error_log_filename}") # 最初の5件のエラーをコンソールにも出力 logger.error("\n【エラー詳細（最初の5件）】") for i, error in enumerate(error_logs[:5], 1): logger.error(f"\n--- エラー {i} ---") logger.error(f"テーブル: {error.get('テーブル名')}") if "検査項目" in error: logger.error(f"検査項目: {error.get('検査項目')}") logger.error(f"エラー: {error.get('エラー内容')}") logger.error(f"SQL: {error.get('SQL')[:100]}..." if len(error.get('SQL', '')) > 100 else f"SQL: {error.get('SQL')}") if len(error_logs) > 5: logger.error(f"\n... 他 {len(error_logs) - 5} 件のエラーがあります。詳細は {error_log_filename} を参照してください。") except Exception as e: logger.error(f"エラーログファイルの出力に失敗しました: {e}") # ファイル出力に失敗してもコンソールには出力する for error in error_logs[:10]: # 最大10件まで logger.error(f"エラー: {error}") def main(): """メイン処理""" # エラーログの初期化 error_logs = [] try: # 変換プロセッサーの初期化 conversion_processor = ConversionProcessor(CONVERSION_RULES_FILE) # 変換ルールの検証 validation_errors = conversion_processor.validate_conversion_rules("T_DONOR_KIDNEY_LIV") if validation_errors: logger.error("変換ルールの検証エラー:") for error in validation_errors: logger.error(f" - {error}") return # デバッグモード：CSVファイルの構造を確認 if logger.level == logging.DEBUG: debug_csv_structure("/csv/liver/shisetsu.csv", sep="\t") debug_csv_structure("/csv/kan/d-followups.csv", encoding="cp932") debug_csv_structure("/csv/kan/r-followups.csv", encoding="cp932") debug_csv_structure("/csv/kan/transplants.csv", encoding="cp932") # データ読み込み institution_df, donor_df, recipient_df, transplants_df, df_mapping = load_data() # データベース接続 connection = pymysql.connect(**DB_CONFIG) with connection.cursor() as cursor: # テーブル初期化 reset_tables(cursor) total_count = len(transplants_df) success_count = 0 # 各移植データの処理 for transplant_index, transplant_raw in transplants_df.iterrows(): logger.info(f"\n処理中: {transplant_index + 1}/{total_count} - YearNo: {transplant_raw.get('YearNo', 'N/A')}") process_transplant_data(cursor, connection, transplant_raw, df_mapping, donor_df, recipient_df, institution_df, error_logs, conversion_processor) success_count += 1 # エラーログの出力 if error_logs: logger.error(f"エラー件数: {len(error_logs)}") for error in error_logs: logger.error(f"エラー詳細: {error}") # 処理結果のサマリー logger.info("\n" + "=" * 50) logger.info("【処理結果サマリー】") logger.info(f"総件数: {total_count}") logger.info(f"成功: {success_count}") logger.info(f"失敗: {total_count - success_count}") logger.info(f"エラー詳細件数: {len(error_logs)}") logger.info("=" * 50 + "\n") # エラーログの出力 output_error_logs(error_logs) except Exception as e: logger.error(f"処理中に予期しないエラーが発生しました: {e}") raise finally: if 'connection' in locals() and connection: connection.close() logger.info("データベース接続を終了しました") if __name__ == "__main__": # 使い方： # 通常実行: python script.py # デバッグモード: DEBUG=1 python script.py main()