""" 腎臓移植データ処理システム（修正版） """ 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 pathlib import Path import time # ConversionProcessorをインポート from conversion_processor import ConversionProcessor # エラーレポート生成モジュールをインポート from error_report_generator import ErrorReportGenerator # ログ設定の初期化 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__) ########################################################################################## ### 設定値（config.pyの内容を統合） ########################################################################################## # データベース接続設定 DB_CONFIG = { "host": "db", "user": "root", "password": "123456", "database": "dev_tracer_db2", "charset": "utf8mb4", "cursorclass": pymysql.cursors.DictCursor, # 接続タイムアウト設定を追加 "connect_timeout": 600, # 接続タイムアウト: 10分 "read_timeout": 600, # 読み取りタイムアウト: 10分 "write_timeout": 600, # 書き込みタイムアウト: 10分 } # ファイルパス設定 CSV_PATHS = { "institution": "/csv/liver/shisetsu.csv", "donor_followups": "/csv/kan/d-followups.csv", "recipient_followups": "/csv/kan/r-followups.csv", "transplants": "/csv/kan/transplants.csv", "mapping_table": "/csv/kan/移行データ対応表.xlsx", "code_mapping": "/csv/kan/移行データ_コード変換表(アルトマーク→TRACER).xlsx", "code_type": "/csv/kan/00_コードタイプ.xlsx", "donor_kidney_liv_conversion_rules": "/csv/kan/donor_kidney_liv_conversion_rules.json", "donor_liv_conversion_rules": "/csv/kan/donor_liv_conversion_rules.json", "recipient_kidney_liv_conversion_rules": "/csv/kan/recipient_kidney_liv_conversion_rules.json", "ishoku_kihon_liv_conversion_rules": "/csv/kan/ishoku_kihon_liv_conversion_rules.json", } # 臓器コードと移植種別コード ORGAN_CODE_KIDNEY = "4" TRANSPLANT_TYPE_LIVING = "1" # バッチ処理の設定 BATCH_SIZE = 100 # 100件ごとに接続を再確立 RETRY_COUNT = 3 # リトライ回数 RETRY_DELAY = 5 # リトライ間隔（秒） # テーブル削除順序（外部キー制約を考慮） 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の診断日"), ] # デバッグ設定 DEBUG_MODE = os.getenv('DEBUG') == '1' DEBUG_BATCH_SIZE = 1 # デバッグ時の処理件数 ########################################################################################## ### データベース接続管理クラス ########################################################################################## class DatabaseConnection: """データベース接続を管理するクラス""" def __init__(self, config: Dict): self.config = config self.connection = None self.connect() def connect(self): """データベースに接続""" try: if self.connection: self.close() self.connection = pymysql.connect(**self.config) logger.info("✅ データベース接続を確立しました") # タイムアウト設定をサーバー側でも設定 with self.connection.cursor() as cursor: cursor.execute("SET SESSION wait_timeout = 28800") # 8時間 cursor.execute("SET SESSION interactive_timeout = 28800") # 8時間 return self.connection except Exception as e: logger.error(f"❌ データベース接続エラー: {e}") raise def reconnect(self): """接続を再確立""" logger.info("🔄 データベース接続を再確立中...") self.close() time.sleep(1) # 短い待機時間 return self.connect() def close(self): """接続を閉じる""" if self.connection: try: self.connection.close() logger.info("✅ データベース接続を終了しました") except: pass self.connection = None def is_connected(self): """接続状態を確認""" if not self.connection: return False try: self.connection.ping(reconnect=False) return True except: return False def ensure_connection(self): """接続を確保（必要に応じて再接続）""" if not self.is_connected(): self.reconnect() return self.connection ########################################################################################## ### リトライ付きデータベース操作 ########################################################################################## def execute_with_retry(db_conn: DatabaseConnection, operation_func: callable, *args, **kwargs): """リトライ付きでデータベース操作を実行""" for attempt in range(RETRY_COUNT): try: # 接続を確保 connection = db_conn.ensure_connection() # 操作を実行 result = operation_func(connection, *args, **kwargs) return result except pymysql.err.OperationalError as e: if "MySQL server has gone away" in str(e) or "Lost connection" in str(e): logger.warning(f"⚠️ 接続エラー (試行 {attempt + 1}/{RETRY_COUNT}): {e}") if attempt < RETRY_COUNT - 1: time.sleep(RETRY_DELAY) db_conn.reconnect() else: raise else: raise except Exception as e: logger.error(f"❌ 予期しないエラー: {e}") raise ########################################################################################## ### 共通更新処理関数 ########################################################################################## def update_table_with_conversion_processor( cursor: pymysql.cursors.DictCursor, connection: pymysql.connections.Connection, table_name: str, primary_key_columns: List[str], primary_key_values: List[Any], conversion_processor: ConversionProcessor, transplant_row: pd.Series, recipient_rows: List[tuple] = None, donor_rows: List[tuple] = None, error_logs: List[Dict] = None ) -> bool: """ ConversionProcessorを使用してテーブルを更新する共通関数 Args: cursor: データベースカーソル connection: データベース接続 table_name: 更新対象テーブル名 primary_key_columns: 主キーカラムのリスト primary_key_values: 主キーの値のリスト conversion_processor: 変換プロセッサ transplant_row: 移植時データ recipient_rows: レシピエントフォローアップデータ donor_rows: ドナーフォローアップデータ error_logs: エラーログリスト Returns: 更新成功の場合True、失敗の場合False """ if error_logs is None: error_logs = [] try: print("#####更新処理#####################################################################################") # 変換ルールを適用してデータを変換 converted_data = conversion_processor.apply_conversion_rules( table_name=table_name, transplant_row=transplant_row, recipient_rows=recipient_rows, donor_rows=donor_rows ) if not converted_data: logger.debug(f"変換データなし: {table_name}") return True # WHERE句の作成 where_conditions = [] where_values = [] for col, val in zip(primary_key_columns, primary_key_values): where_conditions.append(f"`{col}` = %s") where_values.append(val) where_clause = " AND ".join(where_conditions) # 各フィールドを更新 update_count = 0 for column_name, value in converted_data.items(): if value is None: continue try: sql = f"UPDATE {table_name} SET `{column_name}` = %s WHERE {where_clause}" params = [value] + where_values logger.debug(f"実行SQL: {sql} - パラメータ: {params}") cursor.execute(sql, params) update_count += 1 except Exception as e: logger.error(f"カラム更新エラー: {table_name}.{column_name} = {value} - {e}") error_logs.append({ "テーブル名": table_name, "更新カラム名": column_name, "値": value, "エラー内容": str(e), "YearNo": transplant_row.get('YearNo', 'N/A') }) # コミット if update_count > 0: connection.commit() logger.info(f"✅ {table_name} 更新完了: {update_count} カラム") return True except Exception as e: logger.error(f"テーブル更新処理エラー: {table_name} - {e}") connection.rollback() error_logs.append({ "テーブル名": table_name, "エラー内容": str(e), "YearNo": transplant_row.get('YearNo', 'N/A') }) return False ########################################################################################## ### ユーティリティ関数 ########################################################################################## def get_sisetu_cd(facility_name: str, institution_df: pd.DataFrame) -> str: """施設名から施設コードを取得""" if facility_name is None or facility_name == "": return "700000" # カラム名の候補リスト name_columns = ['名称', 'institution_name', '施設名', 'facility_name', 'name'] code_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 code_columns: if col in institution_df.columns: code_col = col break if not name_col or not code_col: logger.warning(f"施設マスタに必要なカラムが見つかりません。デフォルト値を返します。") return "700000" # 完全一致で検索 match = institution_df[institution_df[name_col] == facility_name] if not match.empty: return str(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 str(match_like.iloc[0][code_col]) return "700000" def to_nullable_float(val): """値を浮動小数点数に変換（変換できない場合はNone）""" 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: """追跡調査の種類をソート用の数値に変換""" if "ヶ月" in val: return 0 elif "年" in val: try: return int(val.replace("年後", "").strip()) except: return 99999 return 99999 def to_cycle_str(tsuiseki: str) -> str: """追跡調査の種類をCYCLE文字列に変換""" if not tsuiseki: return "99" tsuiseki = tsuiseki.strip() if tsuiseki == "3ヶ月後": return "99" elif "年後" in tsuiseki: try: year = int(tsuiseki.replace("年後", "").strip()) return f"{year:02d}" except: return "99" return "99" 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 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]) -> Tuple[Optional[int], Optional[str]]: """T_ISHOKU_KIHON_LIVテーブルへの挿入""" columns = ["ISYOKU_ISYOKUSISETU_CD", "ZOKI_CODE", "DEL_FLG", "INS_USER_ID", "INS_PROGRAM_ID"] values = [ transplant_raw.get("ISYOKU_ISYOKUSISETU_CD", "700000"), ORGAN_CODE_KIDNEY, 0, 1, 1 ] seitai_ishoku_id = execute_insert(cursor, connection, "T_ISHOKU_KIHON_LIV", columns, values, error_logs) if seitai_ishoku_id: tracer_id = generate_tracer_id(cursor, transplant_raw.get("移植日", "")) recipient_id = str(seitai_ishoku_id).zfill(7) 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}") return seitai_ishoku_id, tracer_id return None, None def insert_kansen_r_liv(cursor, connection, recipient_row, seitai_ishoku_id, error_logs): """T_KANSEN_R_LIV テーブルへの登録処理""" 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 kansen_map = {"有": 1, "無": 0, "不明": 8} kansen = kansen_map.get(str(kansen_raw).strip()) if kansen is None: continue try: sindan_date = datetime.strptime(str(sindan_date_raw), "%Y/%m/%d").strftime("%Y%m%d") except: sindan_date = None if not sindan_date: continue columns = ["SEITAI_ISYOKU_ID", "KANSEN", "SINDAN_DATE", "DEL_FLG", "INS_USER_ID", "INS_PROGRAM_ID"] values = [seitai_ishoku_id, kansen, sindan_date, "0", "1", "1"] execute_insert(cursor, connection, "T_KANSEN_R_LIV", columns, values, error_logs) 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", "KANJA_ID": seitai_ishoku_id}, {"KANJA_KBN": "1", "KANJA_ID": donor_a_id} ] 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"], "0", "0", "1", "1", "1", "1" ) cursor.execute(sql, values) connection.commit() ########################################################################################## ### データ読み込み関数 ########################################################################################## def load_data(): """データファイルの読み込み""" try: # 施設データ logger.info("施設データを読み込み中...") institution_df = pd.read_csv(CSV_PATHS["institution"], sep="\t", dtype=str).fillna("") logger.info(f"施設データのカラム: {institution_df.columns.tolist()}") # ドナーフォローアップデータ（CP932エンコーディング） logger.info("ドナーフォローアップデータを読み込み中...") donor_df = pd.read_csv(CSV_PATHS["donor_followups"], encoding="cp932", dtype=str).fillna("") logger.info(f"ドナーデータ件数: {len(donor_df)}") # レシピエントフォローアップデータ（CP932エンコーディング） logger.info("レシピエントフォローアップデータを読み込み中...") recipient_df = pd.read_csv(CSV_PATHS["recipient_followups"], encoding="cp932", dtype=str).fillna("") logger.info(f"レシピエントデータ件数: {len(recipient_df)}") # 移植データ（CP932エンコーディング） logger.info("移植データを読み込み中...") transplants_df = pd.read_csv(CSV_PATHS["transplants"], encoding="cp932", dtype=str).fillna("") # デバッグ用：件数制限 if DEBUG_MODE: transplants_df["YearNo_numeric"] = pd.to_numeric(transplants_df["YearNo"], errors="coerce") transplants_df = transplants_df[transplants_df["YearNo_numeric"] >= 20090002] transplants_df = transplants_df.sort_values("YearNo_numeric") transplants_df = transplants_df.head(DEBUG_BATCH_SIZE) 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" # マッピングデータ logger.info("マッピングデータを読み込み中...") mapping_dict = pd.read_excel(CSV_PATHS["mapping_table"], sheet_name=None, header=1) if "腎臓　生体" in mapping_dict: df_mapping = mapping_dict["腎臓　生体"] else: logger.warning("'腎臓　生体'シートが見つかりません。") df_mapping = pd.DataFrame() return institution_df, donor_df, recipient_df, transplants_df, df_mapping except Exception as e: logger.error(f"データ読み込み中にエラーが発生しました: {e}") raise ########################################################################################## ### 各テーブル処理関数 ########################################################################################## def insert_gappei_r_liv(cursor, connection, recipient_rows, seitai_ishoku_id, year_no, error_logs): """T_GAPPEI_R_LIV テーブルへの登録処理""" logger.info("📥 T_GAPPEI_R_LIVへ合併症データを登録中...") for index__, recipient_row in recipient_rows: # 合併症のマッピング gappei_mappings = [ ("合併症等:合併症の有無(調査期間中):糖尿病（ステロイド性を含む）", "03", "0301"), ("合併症等:合併症の有無(調査期間中):高血圧症", "99", "9901"), ("合併症等:合併症の有無(調査期間中):高尿酸血症・痛風", "99", "9901"), ("合併症等:合併症の有無(調査期間中):高脂血症", "99", "9901"), ("合併症等:合併症の有無(調査期間中):虚血性心疾患", "08", "0801"), ("合併症等:合併症の有無(調査期間中):不整脈", "08", "0801"), ("合併症等:合併症の有無(調査期間中):脳血管障害", "09", "0901"), ("合併症等:合併症の有無(調査期間中):その他の心血管系合併症", "08", "0801"), ("合併症等:合併症の有無(調査期間中):肝機能障害", "99", "9901"), ("合併症等:合併症の有無(調査期間中):消化管障害", "99", "9901"), ("合併症等:合併症の有無(調査期間中):骨関節疾患", "99", "9901"), ("合併症等:合併症の有無(調査期間中):悪性腫瘍", "10", "1001"), ("合併症等:合併症の有無(調査期間中):精神障害", "99", "9901") ] for col_name, gappei, gappei_l in gappei_mappings: value = recipient_row.get(col_name) if pd.notna(value) and str(value).strip() == "有": # 診断日を取得 date_col = col_name + "の診断日" sindan_date = recipient_row.get(date_col, "") nyuin_date = "" if pd.notna(sindan_date): try: # 日付フォーマットの変換 date_formats = ["%Y/%m/%d", "%Y-%m-%d", "%Y年%m月%d日"] for fmt in date_formats: try: parsed_date = datetime.strptime(str(sindan_date), fmt) nyuin_date = parsed_date.strftime("%Y%m%d") break except ValueError: continue except Exception as e: logger.warning(f"日付変換エラー: {sindan_date}, {e}") # INSERT処理 columns = [ "SEITAI_ISYOKU_ID", "GAPPEI", "GAPPEI_L", "NYUIN_DATE", "DEL_FLG", "INS_USER_ID", "INS_PROGRAM_ID" ] values = [ seitai_ishoku_id, gappei, gappei_l, nyuin_date, "0", 1, 1 ] execute_insert(cursor, connection, "T_GAPPEI_R_LIV", columns, values, error_logs) ########################################################################################## ### メイン処理関数 ########################################################################################## def process_transplant_data(cursor, connection, transplant_raw, df_mapping, donor_df, recipient_df, institution_df, error_logs, conversion_processors): # 複数形に変更 print(f"@@@@@@@@@@@@@@@@@@@@@ {conversion_processors}") """1件の移植データを処理""" # ConversionProcessorを関数の属性として保存（他の処理で使用するため） #if conversion_processors: # process_transplant_data.conversion_processors = conversion_processors # T_ISHOKU_KIHON_LIVへの挿入 seitai_ishoku_id, tracer_id = insert_ishoku_kihon(cursor, connection, transplant_raw, error_logs) if not seitai_ishoku_id: return year_no = transplant_raw.get('YearNo') # 関連データの取得 recipient_rows = [] donor_rows = [] if year_no: # レシピエントデータ if 'レシピエント追跡調査:YearNo' in recipient_df.columns: 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", ascending=False) recipient_rows = list(recipient_data.iterrows()) # ドナーデータ if 'ドナー追跡調査:YearNo' in donor_df.columns: 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", ascending=False) donor_rows = list(donor_data.iterrows()) # 各テーブルへの処理 donor_a_id = None for target_table in TARGET_TABLES: if target_table == "T_ISHOKU_KIHON_LIV": # 既存のINSERT処理の後に追加 if seitai_ishoku_id and "ishoku_kihon_liv" in conversion_processors: ishoku_conversion_processor = conversion_processors["ishoku_kihon_liv"] update_table_with_conversion_processor( cursor=cursor, connection=connection, table_name=target_table, primary_key_columns=["SEITAI_ISYOKU_ID"], primary_key_values=[seitai_ishoku_id], conversion_processor=ishoku_conversion_processor, transplant_row=transplant_raw, recipient_rows=recipient_rows, donor_rows=donor_rows, error_logs=error_logs ) 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) # T_ISHOKU_KIHON_KIDNEY_LIVの処理部分を修正 elif target_table == 'T_ISHOKU_KIHON_KIDNEY_LIV': columns = ["SEITAI_ISYOKU_ID", "INS_USER_ID", "INS_PROGRAM_ID"] values = [seitai_ishoku_id, 1, 1] kidney_id = execute_insert(cursor, connection, target_table, columns, values, error_logs) # ConversionProcessorを使った更新処理（レシピエント腎臓用） if kidney_id and "recipient_kidney_liv" in conversion_processors: recipient_kidney_conversion_processor = conversion_processors["recipient_kidney_liv"] update_table_with_conversion_processor( cursor=cursor, connection=connection, table_name=target_table, primary_key_columns=["ISYOKU_KIDNEY_ID"], primary_key_values=[kidney_id], conversion_processor=recipient_kidney_conversion_processor, transplant_row=transplant_raw, recipient_rows=recipient_rows, donor_rows=donor_rows, error_logs=error_logs ) elif target_table == 'T_DONOR_LIV': 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) # ConversionProcessorを使った更新処理（ドナー用） print(f"##################################3 {conversion_processors}") if donor_a_id and "donor_liv" in conversion_processors: donor_conversion_processor = conversion_processors["donor_liv"] update_table_with_conversion_processor( cursor=cursor, connection=connection, table_name=target_table, primary_key_columns=["DONOR_A_ID"], primary_key_values=[donor_a_id], conversion_processor=donor_conversion_processor, transplant_row=transplant_raw, recipient_rows=recipient_rows, donor_rows=donor_rows, error_logs=error_logs ) elif target_table == 'T_DONOR_KIDNEY_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] kidney_donor_id = execute_insert(cursor, connection, target_table, columns, values, error_logs) if kidney_donor_id and "donor_kidney_liv" in conversion_processors: donor_kidney_liv_conversion_processor = conversion_processors["donor_kidney_liv"] update_table_with_conversion_processor( cursor=cursor, connection=connection, table_name=target_table, primary_key_columns=["DONOR_A_ID"], primary_key_values=[donor_a_id], conversion_processor=donor_kidney_liv_conversion_processor, transplant_row=transplant_raw, recipient_rows=recipient_rows, donor_rows=donor_rows, error_logs=error_logs ) 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: 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処理 columns = [ "SEITAI_ISYOKU_ID", "GAPPEI", "GAPPEI_L", "NYUIN_DATE", "CMNT", "DEL_FLG", "INS_USER_ID", "INS_PROGRAM_ID" ] values = [ seitai_ishoku_id, "93", gappei_l, nyuin_date, gappei_str[:120], # CMNTは最大120文字 "0", 1, 1 ] execute_insert(cursor, connection, target_table, columns, values, error_logs) logger.debug(f"✅ 合併症{i}を登録: GAPPEI=93, GAPPEI_L={gappei_l}") except Exception as e: logger.error(f"❌ INSERT失敗: T_GAPPEI_R_LIV (合併症{i}), エラー: {e}") error_logs.append({ "テーブル名": "T_GAPPEI_R_LIV", "エラー内容": str(e), "合併症番号": i, "YearNo": year_no }) connection.rollback() logger.info(f"✅ T_GAPPEI_R_LIV登録完了: SEITAI_ISYOKU_ID={seitai_ishoku_id}") 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}") # 免疫抑制薬のマッピング drug_mapping = { "免疫抑制療法:使用薬剤:シクロスポリン": "01", "免疫抑制療法:使用薬剤:タクロリムス": "02", "免疫抑制療法:使用薬剤:アザチオプリン": "03", "免疫抑制療法:使用薬剤:ミゾリビン": "04", "免疫抑制療法:使用薬剤:ミコフェノール酸モフェチル": "05", "免疫抑制療法:使用薬剤:エベロリムス": "06", "免疫抑制療法:使用薬剤:プレドニゾロン": "07", "免疫抑制療法:使用薬剤:メチルプレドニゾロン": "08", "免疫抑制療法:使用薬剤:バシリキシマブ": "11", "免疫抑制療法:使用薬剤:リツキシマブ": "12", "免疫抑制療法:使用薬剤:その他": "99" } # 使用薬剤の確認と登録 drug_found = False for drug_col, drug_code in drug_mapping.items(): drug_value = recipient_row.get(drug_col) # "有"の場合のみ登録 if pd.notna(drug_value) and str(drug_value).strip() == "有": drug_found = True # INSERT処理 columns = [ "SEITAI_ISYOKU_ID", "MENEKI_YOKUSEI", "CYCLE", "DEL_FLG", "INS_USER_ID", "INS_PROGRAM_ID" ] values = [ seitai_ishoku_id, drug_code, cycle_str, "0", 1, 1 ] execute_insert(cursor, connection, target_table, columns, values, error_logs) logger.debug(f"✅ 免疫抑制薬を登録: 薬剤コード={drug_code}, CYCLE={cycle_str}") if drug_found: logger.debug(f"✅ YearNo={year_no}, CYCLE={cycle_str} の免疫抑制薬登録完了") else: logger.debug(f"⚠️ YearNo={year_no}, CYCLE={cycle_str} に使用薬剤なし") 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': # T_LIVING_D_LIVの処理（ドナーの生活状況） if donor_a_id and donor_rows: for _, donor_row in donor_rows: input_date = donor_row.get("ドナー追跡調査:調査日", "") if pd.notna(input_date): try: input_date = pd.to_datetime(input_date).strftime("%Y%m%d") except: continue sisetu_cd = transplant_raw.get("ISYOKU_ISYOKUSISETU_CD", "700000") columns = [ "DONOR_A_ID", "INPUT_DATE", "SISETU_CD", "INS_USER_ID", "INS_PROGRAM_ID" ] values = [donor_a_id, input_date, sisetu_cd, 1, 1] execute_insert(cursor, connection, "T_LIVING_D_LIV", columns, values, error_logs) elif target_table == 'T_KENSA_D_LIV': # ドナー検査データの処理 if donor_a_id and donor_rows: for _, donor_row in donor_rows: # 血清Cr serum_cr = donor_row.get("ドナー:血清Cr") if pd.notna(serum_cr): columns = [ "DONOR_A_ID", "KENSA_NAME", "KENSA_VALUE", "KENSA_UNIT", "INS_USER_ID", "INS_PROGRAM_ID" ] values = [ donor_a_id, "血清Cr", str(serum_cr), "mg/dL", 1, 1 ] execute_insert(cursor, connection, "T_KENSA_D_LIV", columns, values, error_logs) elif target_table == 'T_KENSA_R_LIV': logger.info("📥 T_KENSA_R_LIVへ検査データを登録中...") # レシピエント検査データの処理 for _, recipient_row in recipient_rows: # 追跡調査の種類からCYCLEを決定 tsuiseki_shurui = recipient_row.get("レシピエント追跡調査:追跡調査の種類", "") cycle_str = to_cycle_str(tsuiseki_shurui) if not cycle_str or cycle_str == "99": logger.warning(f"⚠️ CYCLEが不正なためスキップ: {tsuiseki_shurui}") continue # 1. 血清Crの処理 serum_cr = recipient_row.get("移植腎予後:血清Cr") if pd.notna(serum_cr) and str(serum_cr).strip() != "": try: columns = [ "SEITAI_ISYOKU_ID", "KENSA_NAME", "CYCLE", "KENSA_VALUE", "KENSA_UNIT", "KENSA_VALUE_CODE", "KENSA_DATE", "CMNT", "INS_USER_ID", "INS_PROGRAM_ID" ] values = [ seitai_ishoku_id, "血清Cr", cycle_str, str(serum_cr), "mg/dL", None, None, None, 1, 1 ] execute_insert(cursor, connection, target_table, columns, values, error_logs) 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", "エラー内容": str(e) }) connection.rollback() # 2. 尿蛋白/尿中アルブミンの処理 nyo_tanpaku = recipient_row.get("尿検査:尿蛋白") nyo_tanpaku_teiryo = recipient_row.get("尿検査:尿蛋白定量") if pd.notna(nyo_tanpaku) and str(nyo_tanpaku).strip() != "": try: # 尿蛋白のコードマッピング tanpaku_mapping = { "-": "1", "－": "1", "+-": "2", "±": "2", "+": "3", "＋": "3", "++": "4", "＋＋": "4", "+++": "5", "＋＋＋": "5", "++++": "6", "＋＋＋＋": "6" } kensa_value_code = tanpaku_mapping.get(str(nyo_tanpaku).strip(), "9") # 尿蛋白定量値も併記 kensa_value = None if pd.notna(nyo_tanpaku_teiryo) and str(nyo_tanpaku_teiryo).strip() != "": kensa_value = str(nyo_tanpaku_teiryo).strip() columns = [ "SEITAI_ISYOKU_ID", "KENSA_NAME", "CYCLE", "KENSA_VALUE", "KENSA_UNIT", "KENSA_VALUE_CODE", "KENSA_DATE", "CMNT", "INS_USER_ID", "INS_PROGRAM_ID" ] values_tanpaku = [ seitai_ishoku_id, "尿蛋白", cycle_str, kensa_value, "mg/dL" if kensa_value else None, kensa_value_code, None, None, 1, 1 ] execute_insert(cursor, connection, target_table, columns, values_tanpaku, error_logs) logger.debug(f"✅ 尿蛋白を登録: コード={kensa_value_code}, 定量値={kensa_value}") except Exception as e: logger.error(f"❌ INSERT失敗: T_KENSA_R_LIV (尿蛋白), エラー: {e}") error_logs.append({ "テーブル名": "T_KENSA_R_LIV", "検査項目": "尿蛋白", "エラー内容": str(e) }) connection.rollback() # 尿中アルブミンの処理 nyo_albumin = recipient_row.get("尿検査:尿中アルブミン") if pd.notna(nyo_albumin) and str(nyo_albumin).strip() != "": try: # アルブミンのコードマッピング albumin_mapping = { "-": "1", "－": "1", "+-": "2", "±": "2", "+": "3", "＋": "3", "++": "4", "＋＋": "4", "+++": "5", "＋＋＋": "5", "++++": "6", "＋＋＋＋": "6" } kensa_value_code = albumin_mapping.get(str(nyo_albumin).strip(), "9") values_albumin = [ seitai_ishoku_id, "尿中アルブミン", cycle_str, "2", "mg/dL", kensa_value_code, None, None, 1, 1 ] execute_insert(cursor, connection, target_table, columns, values_albumin, error_logs) 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", "検査項目": "尿中蛋白/アルブミン", "エラー内容": 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 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") log_dir = "./logs" os.makedirs(log_dir, exist_ok=True) error_log_filename = f"{log_dir}/migration_errors_{timestamp}.log" try: 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") for key, value in error.items(): f.write(f"{key}: {value}\n") f.write("-" * 50 + "\n\n") logger.info(f"📄 エラー詳細をファイルに出力しました: {error_log_filename}") except Exception as e: logger.error(f"エラーログファイルの出力に失敗しました: {e}") ########################################################################################## ### メイン処理（修正版） ########################################################################################## def main(): """メイン処理""" error_logs = [] db_conn = None conversion_processor = None try: # 入力ファイルの存在確認 for key, path in CSV_PATHS.items(): if key in ["code_mapping", "code_type", "conversion_rules"]: # オプションファイルはスキップ continue if not Path(path).exists(): raise FileNotFoundError(f"必須ファイルが見つかりません: {path}") logger.info("✅ 必須ファイルの存在確認完了") conversion_processors = {} if Path(CSV_PATHS["donor_kidney_liv_conversion_rules"]).exists(): logger.info("🔧 腎臓用ConversionProcessorを初期化中...") conversion_processors["donor_kidney_liv"] = ConversionProcessor( conversion_rules_file=CSV_PATHS["donor_kidney_liv_conversion_rules"], code_mapping_file=CSV_PATHS.get("code_mapping") ) logger.info("✅ 腎臓用ConversionProcessor初期化完了") if Path(CSV_PATHS["donor_liv_conversion_rules"]).exists(): logger.info("🔧 ドナー用ConversionProcessorを初期化中...") conversion_processors["donor_liv"] = ConversionProcessor( conversion_rules_file=CSV_PATHS["donor_liv_conversion_rules"], code_mapping_file=CSV_PATHS.get("code_mapping") ) logger.info("✅ ドナー用ConversionProcessor初期化完了") print(f"CSV: {CSV_PATHS}") if Path(CSV_PATHS["recipient_kidney_liv_conversion_rules"]).exists(): logger.info("🔧 レシピエント腎臓用ConversionProcessorを初期化中...") conversion_processors["recipient_kidney_liv"] = ConversionProcessor( conversion_rules_file=CSV_PATHS["recipient_kidney_liv_conversion_rules"], code_mapping_file=CSV_PATHS.get("code_mapping") ) logger.info("✅ レシピエント腎臓用ConversionProcessor初期化完了") if Path(CSV_PATHS["ishoku_kihon_liv_conversion_rules"]).exists(): logger.info("🔧 T_ISHOKU_KIHON_LIV用ConversionProcessorを初期化中...") conversion_processors["ishoku_kihon_liv"] = ConversionProcessor( conversion_rules_file=CSV_PATHS["ishoku_kihon_liv_conversion_rules"], code_mapping_file=CSV_PATHS.get("code_mapping") ) logger.info("✅ T_ISHOKU_KIHON_LIV用ConversionProcessor初期化完了") if not conversion_processors: logger.warning("⚠️ 変換ルールファイルが見つかりません。ConversionProcessorを使用しません。") # データ読み込み institution_df, donor_df, recipient_df, transplants_df, df_mapping = load_data() # データベース接続管理オブジェクトを作成 db_conn = DatabaseConnection(DB_CONFIG) # テーブル初期化 def reset_tables_with_connection(connection): with connection.cursor() as cursor: reset_tables(cursor) connection.commit() execute_with_retry(db_conn, reset_tables_with_connection) total_count = len(transplants_df) success_count = 0 # バッチ処理の実装 for batch_start in range(0, total_count, BATCH_SIZE): batch_end = min(batch_start + BATCH_SIZE, total_count) batch_df = transplants_df.iloc[batch_start:batch_end] logger.info(f"\n📦 バッチ処理: {batch_start + 1}-{batch_end}/{total_count}") # バッチごとに接続を再確立 if batch_start > 0: db_conn.reconnect() # バッチ内のデータを処理 for idx, (transplant_index, transplant_raw) in enumerate(batch_df.iterrows()): current_index = batch_start + idx + 1 logger.info(f"\n処理中: {current_index}/{total_count} - YearNo: {transplant_raw.get('YearNo', 'N/A')}") try: # リトライ付きで移植データを処理 def process_with_connection(connection, transplant_data): with connection.cursor() as cursor: print(f"################6666666666666 {conversion_processors}") process_transplant_data(cursor, connection, transplant_data, df_mapping, donor_df, recipient_df, institution_df, error_logs, conversion_processors) execute_with_retry(db_conn, process_with_connection, transplant_raw) success_count += 1 except Exception as e: logger.error(f"移植データ処理中にエラー: {e}") error_logs.append({ "テーブル名": "全体処理", "YearNo": transplant_raw.get('YearNo', 'N/A'), "エラー内容": str(e) }) # バッチ処理後の進捗表示 logger.info(f"✅ バッチ完了: {batch_end}/{total_count} ({success_count} 件成功)") # 処理結果のサマリー 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}") import traceback traceback.print_exc() raise finally: if db_conn: db_conn.close() if __name__ == "__main__": main()