""" 腎臓移植データ処理システム（エラーレポート機能付き完全版） このシステムは腎臓移植データをCSVファイルからデータベースに移行するためのプログラムです。 複数の病院からの移植データを統一フォーマットに変換してTRACERデータベースに保存します。 """ # 必要なPythonモジュールをインポートします import os # OS関連の機能（ファイルパスやディレクトリ操作）を使用するため import logging # ログ出力機能を使用してプログラムの動作状況を記録するため import pandas as pd # CSVファイルやExcelファイルを効率的に読み込み・処理するため import pymysql # MySQLデータベースに接続してデータを登録・更新するため from typing import Dict, List, Optional, Tuple, Any # 型ヒントを使ってコードの可読性と保守性を向上させるため from datetime import datetime # 日付や時刻の処理（ログのタイムスタンプなど）を行うため import json # JSONファイル（変換ルールなど）を読み込むため from pathlib import Path # ファイルパスをオブジェクトとして扱い、OS間の違いを吸収するため import time # リトライ処理での待機時間を制御するため import argparse # コマンドライン引数を処理するため # ConversionProcessorをインポート # このモジュールは conversion_map.py の変換ルールを使用してデータ変換を行います from conversion_processor import ConversionProcessor # エラーレポート生成モジュールをインポート # データ移行中に発生したエラーをExcelファイルにまとめて出力するために使用します from error_report_generator import ErrorReportGenerator # ログ設定の初期化 # 環境変数DEBUGが1の場合は詳細なデバッグログを出力し、そうでない場合は通常ログのみ出力します 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の内容を統合） ### システム全体で使用する定数や設定値をここにまとめています ########################################################################################## # データベース接続設定 # MySQLデータベースに接続するための情報を設定しています DB_CONFIG = { "host": "db", # Dockerコンテナ名またはホスト名 "user": "root", # MySQLのユーザー名 "password": "123456", # MySQLのパスワード "database": "dev_TRACERDB6", # 使用するデータベース名 "charset": "utf8mb4", # 文字コード（日本語や絵文字に対応） "cursorclass": pymysql.cursors.DictCursor, # 結果を辞書形式で取得するため # 接続タイムアウト設定を追加 # 長時間の処理で接続が切れないようにするため "connect_timeout": 600, # 接続タイムアウト: 10分 "read_timeout": 600, # 読み取りタイムアウト: 10分 "write_timeout": 600, # 書き込みタイムアウト: 10分 } # ファイルパス設定 # 入力データのCSVファイルやマッピングファイルの場所を指定しています CSV_PATHS = { "institution": "/csv/kan/shisetsu.csv", # 施設マスタ（病院名とコードの対応表） "donor_followups": "/csv/kan/生体腎ドナー移行データ20250812.csv", # ドナーの追跡調査データ "recipient_followups": "/csv/kan/レシピエント追跡移行データ20250819.csv", # レシピエントの追跡調査データ "transplants": "/csv/kan/transplants.csv", # 移植基本情報データ "transplants2": "/csv/kan/basedata20250812差分レコードupload.csv", # 移植基本情報データ "mapping_table": "/csv/kan/移行データ対応表.xlsx", # CSVとDBテーブルのカラム対応表 "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", } # 臓器コードと移植種別コード # TRACER ID生成時に使用する固定値です ORGAN_CODE_KIDNEY = "4" # 腎臓を表すコード TRANSPLANT_TYPE_LIVING = "1" # 生体移植を表すコード # バッチ処理の設定 # 大量データを処理する際に、メモリ使用量や接続の安定性を保つための設定です BATCH_SIZE = 5000 # 100件ごとに接続を再確立してタイムアウトを防ぎます RETRY_COUNT = 3 # リトライ回数 - エラー時に3回まで再試行します RETRY_DELAY = 2 # リトライ間隔（秒） - 再試行前に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', 'T_IJI_MENEKI_YOKUSEI_R_LIV', 'T_KANSEN_R_LIV', 'T_LIVING_D_LIV', #'T_KENSA_D_LIV', 変換表にないので移行対象ではない 'T_KENSA_R_LIV', ] ## 片岡 # 以前の担当者名（変更履歴の記録） # 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を使用してデバッグモードを制御します DEBUG_MODE = os.getenv('DEBUG') == '1' # 環境変数DEBUGが1の場合はデバッグモードON DEBUG_BATCH_SIZE = 1 # デバッグ時の処理件数 - 1件ずつ処理して問題を特定しやすくします ########################################################################################## ### データベース接続管理クラス ### MySQLデータベースへの接続を管理し、安定したデータベース操作を提供します ########################################################################################## class DatabaseConnection: """データベース接続を管理するクラス 接続の確立、再接続、タイムアウト対策などを担当します。 長時間の処理でも接続が切れないように管理しています。 """ def __init__(self, config: Dict): """ コンストラクタ - 接続設定を受け取り、初期接続を確立します """ self.config = config # DB接続設定を保存 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): """接続状態を確認 データベースサーバーに実際にpingして接続が生きているか確認します。 """ if not self.connection: # 接続オブジェクトがない場合 return False try: # pingでサーバーの応答を確認（自動再接続は無効） self.connection.ping(reconnect=False) return True # ping成功 = 接続中 except: return False # ping失敗 = 接続切断 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): """リトライ付きでデータベース操作を実行 ネットワークエラーやタイムアウトで失敗した場合、自動で再試行します。 最大3回までリトライし、それでも失敗した場合はエラーを発生させます。 """ # 設定された回数だけリトライします 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: # MySQL固有の接続エラーをキャッチしてリトライします 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: # その他のOperationalErrorは再試行せずにそのままエラーとします 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, current_csv_type: str = None ) -> bool: """ ConversionProcessorを使用してテーブルを更新する共通関数 """ if error_logs is None: error_logs = [] try: # 変換ルールを適用してデータを変換 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 year_no = transplant_row.get('YearNo', 'N/A') #logger.info(f"\n📝 テーブル更新開始: {table_name} (YearNo: {year_no})") #print(f"\n📝 テーブル更新開始: {table_name} (YearNo: {year_no})") skip_keys = {"target_col", "source_col"} for column_name, value in converted_data.items(): if column_name in skip_keys: continue if value is None: continue try: sql = f"UPDATE {table_name} SET `{column_name}` = %s WHERE {where_clause}" params = [value] + where_values cursor.execute(sql, params) update_count += 1 # 更新成功をログ出力 update_log = ( f" ✅ カラム更新成功:\n" f" CSV: {converted_data["source_col"]}\n" f" テーブル: {table_name}\n" f" カラム: {column_name}\n" f" 設定値: {repr(value)}" ) #logger.info(update_log) #print(f"✅✅✅ DB更新成功: {table_name}.{column_name} = {value}") #logger.info(f"✅✅✅ DB更新成功: {table_name}.{column_name} = {value}") except Exception as e: logger.error(f"カラム更新エラー: {table_name}.{column_name} = {value} - {e}") # エラー情報を詳細に記録 error_logs.append({ "テーブル名": table_name, "更新カラム名": column_name, "値": value, "エラー内容": str(e), "YearNo": year_no, "SQL": sql, "パラメータ": params, "CSV": converted_data["source_col"] or "不明", "主キー値": dict(zip(primary_key_columns, primary_key_values)) }) # コミット if update_count > 0: connection.commit() summary_log = f"✅ {table_name} 更新完了: {update_count} カラム (YearNo: {year_no})" #logger.info(summary_log) #print(f"\n{summary_log}\n") 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'), "CSV": current_csv_type or "不明" }) return False ########################################################################################## ### ユーティリティ関数 ########################################################################################## def get_sisetu_cd(facility_name: str, institution_df: pd.DataFrame) -> str: """施設名から施設コードを取得""" if facility_name is None or facility_name == "": ## テスト病院の施設コードにひもづける return "121212" # 完全一致で検索 match = institution_df[institution_df["JARTRE施設名"] == facility_name] if not match.empty: return str(int(match.iloc[0]["TRACER施設コード"])).zfill(6) else: print(f"{facility_name}はずれ！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！１") # 部分一致で検索 #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 "121212" 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} を初期化しましたaaaaaaaaaaa") except Exception as e: logger.error(f"テーブル {table} の初期化に失敗: {e}") cursor.execute("SET FOREIGN_KEY_CHECKS=1;") logger.info("✅ 外部キー制約を考慮したテーブル初期化完了") ########################################################################################## # 【処理フロー③-4】execute_insert() - DBテーブル登録関数 # 処理内容：指定されたテーブルにINSERT文でデータを登録し、エラー時はログ記録 # 呼び出し関数：なし # 重要処理：INSERT文生成・実行・コミット・エラーハンドリング・自動インクリメントID取得 ########################################################################################## def execute_insert(cursor: pymysql.cursors.DictCursor, connection: pymysql.connections.Connection, table_name: str, columns: List[str], values: List[Any], error_logs: List[Dict], year_no: str = None, current_csv_type: str = None, seitai_isyoku_id: int = None, warning_logs: List[Dict] = None) -> 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}") # 統一フォーマットでエラーログに追加 add_log_entry( log_list=error_logs, level='ERROR', table=table_name, year_no=year_no or "N/A", seitai_isyoku_id=seitai_isyoku_id, message=f"INSERT失敗: {str(e)}", csv_source=current_csv_type or "不明", error_detail=str(e), sql=sql, values=values ) 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", "121212"), ORGAN_CODE_KIDNEY, 0, 1, 1 ] seitai_ishoku_id = execute_insert( cursor, connection, "T_ISHOKU_KIHON_LIV", columns, values, error_logs, year_no=transplant_raw.get('YearNo'), current_csv_type="T_ISHOKU_KIHON_LIV登録処理" ) 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, year_no=None): """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, year_no=year_no, current_csv_type="T_KANSEN_R_LIV登録処理") def add_log_entry(log_list: List[Dict], level: str, table: str, year_no: str, seitai_isyoku_id: int = None, column: str = None, message: str = None, original_value: Any = None, expected_value: Any = None, csv_source: str = None, error_detail: Any = None, sql: str = None, values: List = None) -> None: """統一フォーマットでログエントリを追加 Args: log_list: ログを追加するリスト (error_logs または warning_logs) level: ログレベル ('ERROR' または 'WARNING') table: テーブル名 year_no: YearNo (移植ID) seitai_isyoku_id: 生体移植ID column: 問題のカラム名 message: エラー/警告メッセージ original_value: 元の値 expected_value: 期待される値/形式 csv_source: CSVソース（移植時、ドナー追跡など） error_detail: エラーの詳細（Exceptionオブジェクトなど） sql: 実行されたSQL values: SQLのパラメータ値 """ log_entry = { 'level': level, 'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), 'table': table, 'year_no': year_no, 'seitai_isyoku_id': seitai_isyoku_id, 'column': column, 'message': message, 'original_value': str(original_value) if original_value is not None else None, 'expected_value': str(expected_value) if expected_value is not None else None, 'csv_source': csv_source, 'error_detail': str(error_detail) if error_detail else None, 'sql': sql, 'values': values } log_list.append(log_entry) def insert_nyuryokujokyo(cursor: pymysql.cursors.DictCursor, connection: pymysql.connections.Connection, seitai_ishoku_id: int, donor_a_id: int, error_logs: List[Dict]) -> 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" ) try: cursor.execute(sql, values) connection.commit() except Exception as e: logger.error(f"❌ T_NYURYOKUJOKYO_LIV INSERT失敗: {e}") error_logs.append({ "テーブル名": "T_NYURYOKUJOKYO_LIV", "SQL": sql, "入力値": values, "エラー内容": str(e), "CSV": "移植時" }) ########################################################################################## ### CSV関連のデータの読み込み　データ読み込み関数 ########################################################################################## def _read_csv_auto(path: str, dtype=str) -> pd.DataFrame: """CSVをエンコーディング自動判定で読み込む - 先頭BOMの有無でUTF-8/UTF-16を優先判定 - それ以外は `cp932` → `utf-8-sig` → `utf-8` の順で試行 読み込みに成功したエンコーディングをログ出力する。 """ try_encodings: List[str] = [] try: with open(path, 'rb') as bf: head = bf.read(4) if head.startswith(b"\xef\xbb\xbf"): try_encodings = ["utf-8-sig"] elif head.startswith((b"\xff\xfe", b"\xfe\xff")): try_encodings = ["utf-16"] else: try_encodings = ["cp932", "utf-8-sig", "utf-8"] except Exception: # 万一バイナリ読み込みに失敗してもフォールバックを試す try_encodings = ["cp932", "utf-8-sig", "utf-8"] last_err: Optional[Exception] = None for enc in try_encodings: try: df = pd.read_csv(path, encoding=enc, dtype=dtype).fillna("") logger.info(f"CSV読込成功: {path} (encoding={enc})") return df except Exception as e: last_err = e logger.debug(f"CSV読込失敗: {path} (encoding={enc}) -> {e}") continue # すべて失敗 raise last_err if last_err else RuntimeError(f"CSV読込失敗: {path}") def load_data(): """データファイルの読み込み""" try: # 施設データ logger.info("施設データを読み込み中...") institution_df = pd.read_csv(CSV_PATHS["institution"], dtype=str).fillna("") logger.info(f"施設データのカラム: {institution_df.columns.tolist()}") # ドナーフォローアップデータ（エンコーディング自動判定） logger.info("ドナーフォローアップデータを読み込み中...") donor_df = _read_csv_auto(CSV_PATHS["donor_followups"], dtype=str) logger.info(f"ドナーデータ件数: {len(donor_df)}") # レシピエントフォローアップデータ（エンコーディング自動判定） logger.info("レシピエントフォローアップデータを読み込み中...") recipient_df = _read_csv_auto(CSV_PATHS["recipient_followups"], dtype=str) logger.info(f"レシピエントデータ件数: {len(recipient_df)}") # 移植データ（エンコーディング自動判定） logger.info("移植データを読み込み中...") #### 片岡　CSVの「移植腎の種類」カラムで「生体腎」の値だけ絞り込む transplants_old = _read_csv_auto(CSV_PATHS["transplants"], dtype=str) transplants_new = _read_csv_auto(CSV_PATHS["transplants2"], dtype=str) # それぞれ必要ならフィルタリング def _filter_seitaijin(df: pd.DataFrame) -> pd.DataFrame: if "移植腎の種類" in df.columns: return df[df["移植腎の種類"] == "生体腎"].copy() return df.copy() transplants_old = _filter_seitaijin(transplants_old) transplants_new = _filter_seitaijin(transplants_new) # デバッグ: old/new から YearNo で2件ずつ抜粋してprint if DEBUG_MODE: def _sample_by_yearno(df: pd.DataFrame, n: int = 2) -> pd.DataFrame: if "YearNo" in df.columns: s = pd.to_numeric(df["YearNo"], errors="coerce") tmp = df.copy() tmp["__YearNoNum"] = s tmp = tmp.sort_values(["__YearNoNum", "YearNo"], na_position="last") return tmp.head(n).drop(columns=["__YearNoNum"], errors="ignore") return df.head(n) old_sample = _sample_by_yearno(transplants_old, 2) new_sample = _sample_by_yearno(transplants_new, 2) print("===== DEBUG: old sample (2 rows) =====") try: print(old_sample.to_string(index=False)) except Exception: print(old_sample) print("===== DEBUG: new sample (2 rows) =====") try: print(new_sample.to_string(index=False)) except Exception: print(new_sample) # カラムの和集合で整形し、結合（片方にないカラムは空文字で埋める） union_cols = sorted(set(transplants_old.columns) | set(transplants_new.columns)) transplants_old = transplants_old.reindex(columns=union_cols, fill_value="") transplants_new = transplants_new.reindex(columns=union_cols, fill_value="") transplants_df = pd.concat([transplants_old, transplants_new], ignore_index=True) transplants_df = transplants_df.fillna("") transplants_df.to_csv('to_csv_out.csv') # マージ直後の件数（デバッグ絞り込み適用前） logger.info( f"移植データ件数（マージ直後）: 合計={len(transplants_df)} (old={len(transplants_old)}, new={len(transplants_new)})" ) # デバッグ用：件数制限 片岡（YearNo==240 がある場合のみ絞り込む） if DEBUG_MODE: transplants_df["YearNo_numeric"] = pd.to_numeric(transplants_df.get("YearNo"), errors="coerce") subset240 = transplants_df[transplants_df["YearNo_numeric"] == 19720024] if not subset240.empty: transplants_df = subset240 transplants_df = transplants_df.sort_values("YearNo_numeric") transplants_df = transplants_df.head(max(1, DEBUG_BATCH_SIZE)) logger.info( f"移植データ件数: 合計={len(transplants_df)} (old={len(transplants_old)}, new={len(transplants_new)})" ) logger.info(f"移植データのカラム: {transplants_df.columns.tolist()}") # 施設コード変換 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'] = "121212" # マッピングデータ 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, year_no=year_no, current_csv_type="T_GAPPEI_R_LIV登録処理") ########################################################################################## ### メイン処理関数 ########################################################################################## ########################################################################################## # 【処理フロー③】process_transplant_data() - 1件の移植データ処理関数 # 処理内容：1件の移植データに対して、関連データ取得・ソート・各テーブル確定処理 # 呼び出し関数：③-1 insert_ishoku_kihon() → ③-2 parse_followup_cycle() → ③-3 update_table_with_conversion_processor() → ③-4 execute_insert() # 重要処理：YearNoで関連データ取得・経過年数降順ソート・10テーブルへのデータ登録更新 ########################################################################################## def process_transplant_data(cursor, connection, transplant_raw, df_mapping, donor_df, recipient_df, institution_df, error_logs, conversion_processors): """1件の移植データを処理""" ########################################################################################## # 【処理ステップ①】T_ISHOKU_KIHON_LIVテーブルへの移植基本情報登録 ########################################################################################## seitai_ishoku_id, tracer_id = insert_ishoku_kihon(cursor, connection, transplant_raw, error_logs) # ③-1呼び出し if not seitai_ishoku_id: # 登録に失敗した場合は処理を中断 return year_no = transplant_raw.get('YearNo') ########################################################################################## # 【処理ステップ②】関連データの取得とソート処理 # ここが最も重要：YearNoをキーに関連するレシピエント・ドナーの追跡データを取得し、 # 経過年数の降順（7年後→6年後→...→1年後→3ヶ月後）でソートする ########################################################################################## recipient_rows = [] # レシピエント追跡データを格納するリスト donor_rows = [] # ドナー追跡データを格納するリスト if year_no: # YearNoが存在する場合のみ処理 # 【②-1】レシピエント追跡データの取得・ソート処理 if 'レシピエント追跡調査:YearNo' in recipient_df.columns: # YearNoで一致するレシピエント追跡データを全て抽出 # 例：YearNo="20120768"に対応する全ての追跡調査データ（3ヶ月後、1年後、3年後、5年後など） recipient_data = recipient_df[recipient_df['レシピエント追跡調査:YearNo'] == year_no].copy() # 追跡調査の種類でソート処理を実行 if 'レシピエント追跡調査:追跡調査の種類' in recipient_data.columns: # parse_followup_cycle関数で文字列を数値に変換（"5年後"→5, "3ヶ月後"→0） recipient_data.loc[:, "sort_key"] = recipient_data["レシピエント追跡調査:追跡調査の種類"].apply(parse_followup_cycle) # 降順ソート実行：大きい数値が先頭（7年後→6年後→...→1年後→3ヶ月後） # これにより最も長期間の経過観察データが最優先で取得される recipient_data = recipient_data.sort_values("sort_key", ascending=False) # pandas DataFrameをリスト形式に変換（後続処理で使いやすくするため） recipient_rows = list(recipient_data.iterrows()) # タプル(index, Series)のリストに変換して処理しやすくする # 【②-2】ドナー追跡データの取得・ソート処理（レシピエントと同じロジック） if 'ドナー追跡調査:YearNo' in donor_df.columns: # ドナー追跡データに YearNo カラムが存在するかチェック # YearNoで一致するドナー追跡データを全て抽出 donor_data = donor_df[donor_df['ドナー追跡調査:YearNo'] == year_no].copy() # 現在処理中の移植データのYearNoと一致するドナー追跡データを抽出 # 追跡調査の種類でソート処理を実行 if 'ドナー追跡調査:追跡調査の種類' in donor_data.columns: # 追跡調査の種類カラムが存在するかチェック # parse_followup_cycle関数で文字列を数値に変換 donor_data.loc[:, "sort_key"] = donor_data["ドナー追跡調査:追跡調査の種類"].apply(parse_followup_cycle) # "３ヶ月後"=0, "1年後"=1 のように数値化してソート用キーを作成 # 降順ソート実行：最も長期間の経過観察データが最優先 donor_data = donor_data.sort_values("sort_key", ascending=False) # 降順ソートで「7年後」→「6年後」→...→「3ヶ月後」の順番に並び替え # pandas DataFrameをリスト形式に変換 donor_rows = list(donor_data.iterrows()) # タプル(index, Series)のリストに変換して処理しやすくする ########################################################################################## # 【処理ステップ③】各テーブルへの順次データ登録・更新処理 # 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’, # ‘T_IJI_MENEKI_YOKUSEI_R_LIV’, ‘T_KANSEN_R_LIV’, ‘T_LIVING_D_LIV’, # ‘T_KENSA_D_LIV’, ‘T_KENSA_R_LIV’] ########################################################################################## donor_a_id = None # ドナーA_IDを保存する変数（後続テーブルで使用） # 各テーブルを順番に処理（外部キー関係を考慮した順序） for target_table in TARGET_TABLES: ########################################################################################## # 【テーブル処理③-①】T_ISHOKU_KIHON_LIV - 移植基本テーブルのconversion_map変換更新 ########################################################################################## if target_table == "T_ISHOKU_KIHON_LIV": # 移植基本情報テーブルの場合 #logger.info(f"🔍 T_ISHOKU_KIHON_LIV処理開始") #logger.info(f"🔍 seitai_ishoku_id: {seitai_ishoku_id}") #logger.info(f"🔍 conversion_processors keys: {conversion_processors.keys()}") if seitai_ishoku_id and "ishoku_kihon_liv" in conversion_processors: #logger.info(f"✅ T_ISHOKU_KIHON_LIV用ConversionProcessorが見つかりました") ishoku_conversion_processor = conversion_processors["ishoku_kihon_liv"] # 変換ルールの内容を確認 table_rules = ishoku_conversion_processor.get_table_rules("T_ISHOKU_KIHON_LIV") #logger.info(f"🔍 T_ISHOKU_KIHON_LIVの変換ルール数: {len(table_rules)}") 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, current_csv_type="T_ISHOKU_KIHON_LIV 更新処理" ) else: logger.warning(f"⚠️ T_ISHOKU_KIHON_LIV用ConversionProcessorが見つかりません") if not seitai_ishoku_id: logger.warning(f"⚠️ seitai_ishoku_idが存在しません") ########################################################################################## # 【テーブル処理③-②】T_TRACER_IKO - TRACER移行テーブルの基本情報登録 ########################################################################################## elif target_table == 'T_TRACER_IKO': # TRACER移行情報テーブルの場合 # TRACER移行テーブルに固定値を登録（変換処理なし） columns = ["TRACER_ID", "ZOKI_CODE", "SEITAI_ISYOKU_ID", "ISHOKU_TOROKU_ID"] # 登録カラム定義 values = [tracer_id, 4, seitai_ishoku_id, year_no] # 登録値定義（ZOKI_CODE=4は腎臓コード） execute_insert(cursor, connection, target_table, columns, values, error_logs, # ③-4呼び出し year_no=year_no, current_csv_type="移植時") ########################################################################################## # 【テーブル処理③-③】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, year_no=year_no, current_csv_type="移植時") # ConversionProcessorを使った更新処理（レシピエント腎臓用） if kidney_id and "recipient_kidney_liv" in conversion_processors: recipient_kidney_conversion_processor = conversion_processors["recipient_kidney_liv"] #print("##########################################################################################") 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, current_csv_type="T_ISHOKU_KIHON_KIDNEY_LIV 更新処理" ) # 免疫抑制剤（カルシニュリン・インヒビター）のフラグ設定処理 if kidney_id: #logger.info("🔬 免疫抑制剤フラグ設定処理開始") # CSVから免疫抑制剤の値を取得 calcineurin_inhibitor = transplant_raw.get("免疫抑制剤(導入期):カルシニュリン・インヒビター", "") # デフォルト値を設定 donyu_csa = "0" donyu_tac = "0" tac_type = "0" # 値に応じてフラグを設定 if pd.notna(calcineurin_inhibitor) and str(calcineurin_inhibitor).strip(): value = str(calcineurin_inhibitor).strip() logger.debug(f"免疫抑制剤の値: {value}") # シクロスポリンのチェック if "シクロスポリン(CyA)" in value or "シクロスポリン" in value: donyu_csa = "1" #logger.info("✅ シクロスポリン(CyA)を検出 → DONYU_CSA=1") # タクロリムスのチェック if "タクロリムス(FK506)" in value or "タクロリムス-ER" in value or "タクロリムス" in value: donyu_tac = "1" tac_type = "1" #logger.info("✅ タクロリムスを検出 → DONYU_TAC=1, TAC_TYPE=1") # 複数の薬剤が含まれる場合の処理（カンマ区切り対応） if "," in value or "、" in value: #logger.info("🔄 複数の薬剤を検出") # カンマまたは読点で分割 values = value.replace("、", ",").split(",") for v in values: v = v.strip() if "シクロスポリン" in v: donyu_csa = "1" if "タクロリムス" in v: donyu_tac = "1" tac_type = "1" # データベースを更新 try: update_sql = f""" UPDATE T_ISHOKU_KIHON_KIDNEY_LIV SET DONYU_CSA = %s, DONYU_TAC = %s, TAC_TYPE = %s, UPD_USER_ID = 1, UPD_PROGRAM_ID = 1, UPD_DATE = NOW() WHERE ISYOKU_KIDNEY_ID = %s """ cursor.execute(update_sql, [donyu_csa, donyu_tac, tac_type, kidney_id]) connection.commit() #logger.info(f"✅ 免疫抑制剤フラグ更新成功: DONYU_CSA={donyu_csa}, DONYU_TAC={donyu_tac}, TAC_TYPE={tac_type}") except Exception as e: logger.error(f"❌ 免疫抑制剤フラグ更新エラー: {e}") error_logs.append({ "テーブル名": "T_ISHOKU_KIHON_KIDNEY_LIV", "エラー内容": f"免疫抑制剤フラグ更新失敗: {str(e)}", "YearNo": year_no, "カルシニュリン・インヒビター": calcineurin_inhibitor }) 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, year_no=year_no, current_csv_type="移植時") # ConversionProcessorを使った更新処理（ドナー用） 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, current_csv_type="T_DONOR_LIV更新処理" ) 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] donor_kidney_id = execute_insert(cursor, connection, target_table, columns, values, error_logs, year_no=year_no, current_csv_type="移植時") #print("🚨🚨🚨 T_DONOR_KIDNEY_LIV処理開始 🚨🚨🚨") #print(f"🔍 donor_kidney_id: {donor_kidney_id}") #print(f"🔍 利用可能なconversion_processors: {list(conversion_processors.keys())}") # ConversionProcessorを使った更新処理（ドナー腎臓用） if donor_kidney_id and "donor_kidney_liv" in conversion_processors: #print("✅ donor_kidney_liv ConversionProcessorが見つかりました") donor_kidney_conversion_processor = conversion_processors["donor_kidney_liv"] # 🚨 デバッグ用: 変換ルールの中身を確認 table_rules = donor_kidney_conversion_processor.get_table_rules("T_DONOR_KIDNEY_LIV") #print(f"🔍 T_DONOR_KIDNEY_LIVの変換ルール数: {len(table_rules)}") #print(f"🔍 変換ルール一覧: {list(table_rules.keys())}") if "KOKETSUATSU" in table_rules: #print("🎯 KOKETSUATSUルールを発見！") koketsuatsu_rule = table_rules["KOKETSUATSU"] #print(f"🎯 KOKETSUATSUルール詳細: {koketsuatsu_rule}") #print(f"🎯 KOKETSUATSUタイプ: {koketsuatsu_rule.get('type')}") try: # まず変換処理をテスト実行 test_converted_data = donor_kidney_conversion_processor.apply_conversion_rules( table_name="T_DONOR_KIDNEY_LIV", transplant_row=transplant_raw, recipient_rows=recipient_rows, donor_rows=donor_rows ) update_success = update_table_with_conversion_processor( cursor=cursor, connection=connection, table_name=target_table, primary_key_columns=["DONOR_KIDNEY_ID"], primary_key_values=[donor_kidney_id], conversion_processor=donor_kidney_conversion_processor, transplant_row=transplant_raw, recipient_rows=recipient_rows, donor_rows=donor_rows, error_logs=error_logs, current_csv_type="T_DONOR_KIDNEY_LIV更新処理" ) if update_success: # 🚨 実際にDBが更新されたかを確認 cursor.execute("SELECT KOKETSUATSU FROM T_DONOR_KIDNEY_LIV WHERE DONOR_KIDNEY_ID = %s", (donor_kidney_id,)) db_result = cursor.fetchone() except Exception as update_error: logger.error(f"T_DONOR_KIDNEY_LIV Update処理エラー: {update_error}") error_logs.append({ "テーブル名": "T_DONOR_KIDNEY_LIV", "エラー内容": str(update_error), "処理段階": "ConversionProcessor更新", "YearNo": year_no }) 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, year_no=year_no, current_csv_type="レシピエント追跡") 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, "CSV": "レシピエント追跡" }) 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": continue logger.debug(f"処理中: YearNo={year_no}, 追跡調査種類={tsuiseki_shurui}, CYCLE={cycle_str}") # 免疫抑制薬のフラグ情報を初期化 drug_flags = { "CS": 0, "CSA": 0, "TAC": 0, "Rap": 0, "AZ": 0, "CP": 0, "MMF": 0, "MZ": 0, "MENEKI_ETC": 0, "MENEKI_ETC_CMNT": "" } # ステロイド steroid_value = recipient_row.get("免疫抑制剤:ステロイド", "") if pd.notna(steroid_value) and str(steroid_value).strip() == "使用": drug_flags["CS"] = 1 # カルシニュリン・インヒビター cni_value = recipient_row.get("免疫抑制剤:カルシニュリン・インヒビター", "") if pd.notna(cni_value): cni_str = str(cni_value).strip() if cni_str in ["シクロスポリン(CyA)", "シクロスポリン（CyA）"]: drug_flags["CSA"] = 1 elif cni_str in ["タクロリムス(FK506)", "タクロリムス（FK506）", "タクロリムス-ER"]: drug_flags["TAC"] = 1 # mTOR阻害剤 mtor_value = recipient_row.get("免疫抑制剤:mTOR阻害剤", "") if pd.notna(mtor_value): mtor_str = str(mtor_value).strip() if mtor_str in ["エベロリムス(RAD)", "エベロリムス（RAD）", "シロリムス(Rapamycin)", "シロリムス（Rapamycin）"]: drug_flags["Rap"] = 1 # 核酸合成阻害剤（カンマ区切りで処理） kakusan_value = recipient_row.get("免疫抑制剤:核酸合成阻害剤", "") if pd.notna(kakusan_value): kakusan_str = str(kakusan_value).strip() # カンマ区切りで分割 kakusan_items = [item.strip() for item in kakusan_str.split(",")] for item in kakusan_items: if item in ["アザチオプリン(AZP)", "アザチオプリン（AZP）"]: drug_flags["AZ"] = 1 elif item == "シクロホスファミド(CP)": drug_flags["CP"] = 1 elif item == "ミコフェノール酸モフェチル(MMF)": drug_flags["MMF"] = 1 elif item in ["ミゾリビン(MZR)", "ミゾリビン（MZR）"]: drug_flags["MZ"] = 1 # その他の使用 etc_usage_value = recipient_row.get("免疫抑制剤:その他の使用", "") if pd.notna(etc_usage_value) and str(etc_usage_value).strip() == "使用": drug_flags["MENEKI_ETC"] = 1 # 薬剤名（そのまま設定） drug_name_value = recipient_row.get("免疫抑制剤:薬剤名", "") if pd.notna(drug_name_value): drug_flags["MENEKI_ETC_CMNT"] = str(drug_name_value) # いずれかの免疫抑制薬が使用されている場合のみ登録 if any(value == 1 for key, value in drug_flags.items() if key != "MENEKI_ETC_CMNT"): # INSERT処理 columns = [ "SEITAI_ISYOKU_ID", "CYCLE", "CS", "CSA", "TAC", "Rap", "AZ", "CP", "MMF", "MZ", "MENEKI_ETC", "MENEKI_ETC_CMNT", "INS_USER_ID", "INS_PROGRAM_ID" ] values = [ seitai_ishoku_id, cycle_str, drug_flags["CS"], drug_flags["CSA"], drug_flags["TAC"], drug_flags["Rap"], drug_flags["AZ"], drug_flags["CP"], drug_flags["MMF"], drug_flags["MZ"], drug_flags["MENEKI_ETC"], drug_flags["MENEKI_ETC_CMNT"], 1, 1 ] execute_insert(cursor, connection, target_table, columns, values, error_logs, year_no=year_no, current_csv_type="レシピエント追跡") logger.debug(f"✅ 免疫抑制薬を登録: CYCLE={cycle_str}, フラグ={drug_flags}") 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, year_no) elif target_table == 'T_LIVING_D_LIV': # T_LIVING_D_LIVの処理（ドナーの生活状況） if donor_a_id and donor_rows: #logger.info("📥 T_LIVING_D_LIVへドナー生活状況データを登録中...") 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 #### #### 施設コードのCSVを修正したときに下記の処理を見直す #### sisetu_cd = transplant_raw.get("ISYOKU_ISYOKUSISETU_CD", "121212") # 身長・体重・血圧の取得 height = donor_row.get("ドナー所見:身長", None) if pd.notna(height) and str(height).strip() != "": height = str(height).strip() else: height = None weight = donor_row.get("ドナー所見:体重", None) if pd.notna(weight) and str(weight).strip() != "": weight = str(weight).strip() else: weight = None ketsuatsu_upper = donor_row.get("ドナー所見:収縮期", "") if pd.notna(ketsuatsu_upper): ketsuatsu_upper = str(ketsuatsu_upper).strip() else: ketsuatsu_upper = None ketsuatsu_lower = donor_row.get("ドナー所見:拡張期", "") if pd.notna(ketsuatsu_lower): ketsuatsu_lower = str(ketsuatsu_lower).strip() else: ketsuatsu_lower = None # 社会復帰状況:精神的の処理 shakai_seshin_value = donor_row.get("ドナー所見:社会復帰状況:精神的", "") shakai_seshin = None if pd.notna(shakai_seshin_value): seshin_str = str(shakai_seshin_value).strip() if seshin_str == "良好": shakai_seshin = 1 elif seshin_str == "変化なし": shakai_seshin = 2 elif seshin_str == "不良": shakai_seshin = 3 elif seshin_str == "不明": shakai_seshin = 4 # 社会復帰状況:身体的の処理 shakai_shintai_value = donor_row.get("ドナー所見:社会復帰状況:身体的", "") shakai_shintai = None if pd.notna(shakai_shintai_value): shintai_str = str(shakai_shintai_value).strip() if shintai_str == "良好": shakai_shintai = 1 elif shintai_str == "変化なし": shakai_shintai = 2 elif shintai_str == "不良": shakai_shintai = 3 elif shintai_str == "不明": shakai_shintai = 4 columns = [ "DONOR_A_ID", "INPUT_DATE", "SISETU_CD", "HEIGHT", "WEIGHT", "KETSUKATSU_UPPER", "KETSUKATSU_LOWER", "SHAKAIFUKKI_JOKYO_SESHIN", "SHAKAIFUKKI_JOKYO_SHINTAI", "INS_USER_ID", "INS_PROGRAM_ID" ] values = [ donor_a_id, input_date, sisetu_cd, height, weight, ketsuatsu_upper, ketsuatsu_lower, shakai_seshin, shakai_shintai, 1, 1 ] execute_insert(cursor, connection, "T_LIVING_D_LIV", columns, values, error_logs, year_no=year_no, current_csv_type="ドナー追跡") logger.debug(f"✅ ドナー生活状況データを登録: HEIGHT={height}, WEIGHT={weight}, " f"BP={ketsuatsu_upper}/{ketsuatsu_lower}, 精神的={shakai_seshin}, 身体的={shakai_shintai}") #logger.info(f"✅ T_LIVING_D_LIV登録完了: DONOR_A_ID={donor_a_id}") 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) if not cycle_str or cycle_str == "99": continue 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}") 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": 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, year_no=year_no, current_csv_type="レシピエント追跡") 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), "CSV": "レシピエント追跡" }) 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, year_no=year_no, current_csv_type="レシピエント追跡") 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), "CSV": "レシピエント追跡" }) 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, year_no=year_no, current_csv_type="レシピエント追跡") 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), "CSV": "レシピエント追跡" }) 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, error_logs) ########################################################################################## # 【処理フロー④】output_error_logs() - エラーレポート生成関数 # 処理内容：処理中に発生したエラー情報をCSV・Excelファイルとして出力し、統計情報を表示 # 呼び出し関数：④-1 ErrorReportGenerator.generate_detailed_error_report() (エラーレポート生成) # 重要処理：エラー統計、テーブル別エラー分析、時間スタンプ付きファイル出力 ########################################################################################## def output_error_logs(error_logs: List[Dict], warning_logs: List[Dict], transplant_df: pd.DataFrame, df_mapping: pd.DataFrame, db_conn: DatabaseConnection, start_index: int = 0, end_index: int = None) -> None: """エラーと警告ログを統一フォーマットで出力し、詳細レポートを生成（分割実行対応）""" # エラーと警告の件数を確認 has_errors = len(error_logs) > 0 has_warnings = len(warning_logs) > 0 total_logs = len(error_logs) + len(warning_logs) if has_errors: 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}件") #else: #logger.info("✅ エラーなく処理が完了しました（ログファイルは作成します）") # エラー詳細をログファイルに出力 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") log_dir = "./logs" os.makedirs(log_dir, exist_ok=True) # 旧形式のエラーログも出力（互換性のため）- 分割実行対応 if end_index is not None: range_suffix = f"_{start_index + 1:06d}-{end_index:06d}" else: range_suffix = f"_{start_index + 1:06d}-end" error_log_filename = f"{log_dir}/migration_errors_{timestamp}{range_suffix}.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"処理範囲: {start_index + 1}-{end_index if end_index else 'end'}\n") f.write(f"エラー総数: {len(error_logs)}件\n") f.write(f"警告総数: {len(warning_logs)}件\n") f.write("=" * 50 + "\n\n") # エラーログの出力 if has_errors: f.write("【エラーログ】\n") f.write("=" * 50 + "\n") for i, error in enumerate(error_logs, 1): f.write(f"[ERROR {i}/{len(error_logs)}] ") f.write(f"YearNo:{error.get('year_no', 'N/A')} | ") f.write(f"{error.get('table', 'N/A')}") if error.get('column'): f.write(f".{error['column']}") f.write(f" | {error.get('message', 'N/A')}\n") if error.get('original_value'): f.write(f" 元の値: {error['original_value']}\n") if error.get('expected_value'): f.write(f" 期待値: {error['expected_value']}\n") if error.get('error_detail'): f.write(f" 詳細: {error['error_detail']}\n") f.write("-" * 50 + "\n") # 警告ログの出力 if has_warnings: f.write("\n【警告ログ】\n") f.write("=" * 50 + "\n") for i, warning in enumerate(warning_logs, 1): f.write(f"[WARNING {i}/{len(warning_logs)}] ") f.write(f"YearNo:{warning.get('year_no', 'N/A')} | ") f.write(f"{warning.get('table', 'N/A')}") if warning.get('column'): f.write(f".{warning['column']}") f.write(f" | {warning.get('message', 'N/A')}\n") if warning.get('original_value'): f.write(f" 元の値: {warning['original_value']}\n") if warning.get('expected_value'): f.write(f" 期待値: {warning['expected_value']}\n") f.write("-" * 50 + "\n") if not has_errors and not has_warnings: f.write("エラー・警告は発生しませんでした。処理は正常に完了しました。\n\n") f.write("この範囲の処理では以下が実行されました:\n") f.write(f"- 処理開始位置: {start_index + 1}\n") f.write(f"- 処理終了位置: {end_index if end_index else '最後まで'}\n") f.write(f"- 全体の件数に対する範囲: {start_index + 1}-{end_index if end_index else 'end'}\n") #logger.info(f"📄 エラーログファイルを作成しました: {error_log_filename}") except Exception as e: logger.error(f"エラーログファイルの作成に失敗しました: {e}") # ErrorReportGeneratorを使用した詳細レポート生成 if db_conn and db_conn.is_connected(): try: logger.info("📊 詳細エラーレポートを生成中...") error_report_generator = ErrorReportGenerator(db_conn.connection) # エラーと警告を統合したリストを作成 all_logs = error_logs + warning_logs # エラーレポートの生成（エラーと警告を含む） error_report_df = error_report_generator.generate_error_report( error_logs=all_logs, transplant_df=transplant_df, mapping_df=df_mapping ) # レポートの保存（分割実行対応） excel_path = error_report_generator.save_error_report(error_report_df, log_dir, range_suffix=range_suffix) # サマリーレポートの生成 summary_df = error_report_generator.generate_summary_report(error_report_df) # サマリーを画面に表示 logger.info("\n【エラーサマリー】") for _, row in summary_df.iterrows(): # 統一フォーマットの場合は'項目'列を使用、従来形式は'エラータイプ'列を使用 if '項目' in row and row['項目']: logger.info(f"- {row['分類']}: {row['項目']}: {row['件数']}件") elif 'エラータイプ' in row: logger.info(f"- {row['エラータイプ']}: {row['件数']}件") else: logger.info(f"- 不明: {row['件数']}件") if '主な対処方法' in row and row['主な対処方法'] != '-': logger.info(f" 対処方法: {row['主な対処方法']}") logger.info(f"\n✅ 詳細エラーレポートを保存しました: {excel_path}") except Exception as e: logger.error(f"詳細エラーレポートの生成に失敗しました: {e}") ########################################################################################## ### メイン処理（修正版） ########################################################################################## def main(start_index: int = 0, limit: int = None, clear_tables: bool = True): # プログラムのメインエントリーポイント関数 """メイン処理 この関数は、データ移行プログラムの主処理を行います。 CSVファイルからデータを読み込み、変換してデータベースに登録します。 Args: start_index: 処理開始インデックス（0から開始） limit: 処理件数制限（None の場合は全件処理） clear_tables: テーブルクリア実行フラグ（Falseの場合はテーブルクリアをスキップ） """ error_logs = [] # エラー情報を記録するリストを初期化 warning_logs = [] # 警告情報を記録するリストを初期化 db_conn = None # データベース接続オブジェクトを初期化 conversion_processor = None # データ変換プロセッサオブジェクトを初期化 try: # エラーハンドリングを行うtryブロック開始 # ConversionProcessorの共有CSVをリセット（新しいセッション開始） # 前回の実行でキャッシュされたデータをクリアして新しい処理を開始します ConversionProcessor.reset_shared_csv() # 入力ファイルの存在確認 #### [AIへ] 回答: JSONやExcelファイルはオプションアルで、必須ではないためスキップしています。CSVファイルのみ必須チェックです。 # 処理を始める前に、必要なCSVファイルがすべて存在するかチェックします for key, path in CSV_PATHS.items(): # 設定された全ファイルパスをループでチェック ## JSON及びExcelファイルはスキップする if key in ["code_mapping", "code_type", "conversion_rules"]: # オプションファイルはスキップ continue # オプションファイルの場合はチェックを飛ばして次のファイルへ if not Path(path).exists(): # ファイルが存在しない場合 raise FileNotFoundError(f"必須ファイルが見つかりません: {path}") # エラーを発生させて処理を停止 #logger.info("✅ 必須ファイルの存在確認完了") # ファイル確認完了をログ出力 ##　移植時及びフォロー及び施設CSVファイルの読み込み 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") # コードマッピングファイルを指定（オプション） ) # ConversionProcessorコンストラクタ終了 #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") # コードマッピングファイルを指定（オプション） ) # ConversionProcessorコンストラクタ終了 #logger.info("✅ ドナー用ConversionProcessor初期化完了") # 初期化完了をログ出力 # レシピエント腎臓用のデータ変換ルールファイルが存在するかチェック 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") # コードマッピングファイルを指定（オプション） ) # ConversionProcessorコンストラクタ終了 #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") # コードマッピングファイルを指定（オプション） ) # ConversionProcessorコンストラクタ終了 #logger.info("✅ T_ISHOKU_KIHON_LIV用ConversionProcessor初期化完了") # 初期化完了をログ出力 if not conversion_processors: logger.warning("⚠️ 変換ルールファイルが見つかりません。ConversionProcessorを使用しません。") # CSVファイルのデータ読み込み institution_df, donor_df, recipient_df, transplants_df, df_mapping = load_data() if conversion_processors: #logger.info("📁 ConversionProcessorにファイルデータを読み込み中...") for name, processor in conversion_processors.items(): # 各プロセッサーにファイルデータを読み込み #### [AIへ] 回答: はい、その通りです。load_dataで読み込んだCSVデータをConversionProcessorインスタンスに渡しています。 processor.load_file_data("transplants", transplants_df) processor.load_file_data("d-followups", donor_df) processor.load_file_data("r-followups", recipient_df) #logger.info(f"✅ {name}: ファイルデータ読み込み完了") # データベース接続管理オブジェクトを作成 db_conn = DatabaseConnection(DB_CONFIG) # テーブル初期化（start_index=0の場合のみ実行） if clear_tables: #logger.info("🗑️ テーブル初期化を実行中...") 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) #logger.info("✅ テーブル初期化完了") else: logger.info("⚠️ テーブル初期化をスキップしました") total_count = len(transplants_df) # 分割実行の範囲を調整 if limit is not None: end_index = min(start_index + limit, total_count) else: end_index = total_count # 処理対象のデータを絞り込み process_df = transplants_df.iloc[start_index:end_index] process_count = len(process_df) success_count = 0 #logger.info(f"📊 処理範囲: {start_index + 1}-{end_index}/{total_count} (処理対象: {process_count}件)") # バッチ処理の実装（処理対象データに対してバッチ処理） for batch_start in range(0, process_count, BATCH_SIZE): batch_end = min(batch_start + BATCH_SIZE, process_count) #### [AIへ] 回答: はい、ここでtransplants.csvのバッチデータを取得しています。100件ずつ処理します。 batch_df = process_df.iloc[batch_start:batch_end] actual_start = start_index + batch_start + 1 actual_end = start_index + batch_end #logger.info(f"\n📦 バッチ処理: {actual_start}-{actual_end}/{total_count}") # バッチごとに接続を再確立 if batch_start > 0: db_conn.reconnect() # バッチ内のデータを処理 for idx, (transplant_index, transplant_raw) in enumerate(batch_df.iterrows()): current_index = start_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: process_transplant_data(cursor, connection, transplant_data, df_mapping, donor_df, recipient_df, institution_df, error_logs, conversion_processors) #### [AIへ] 回答: はい、その通りです。execute_with_retry関数でDB接続エラー時の自動リトライ処理を行っています。 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"処理範囲: {start_index + 1}-{end_index}") logger.info(f"処理対象件数: {process_count}") logger.info(f"成功: {success_count}") logger.info(f"失敗: {process_count - success_count}") logger.info(f"エラー詳細件数: {len(error_logs)}") logger.info("=" * 50 + "\n") # 統合変換ログのサマリーを保存 if conversion_processors: ConversionProcessor.save_global_conversion_summary(conversion_processors) csv_filename = ConversionProcessor.get_shared_csv_filename() if csv_filename: logger.info(f"✅ 統合変換ログを保存しました: {csv_filename}") # エラーログの出力（ErrorReportGenerator付き） output_error_logs(error_logs, warning_logs, transplants_df, df_mapping, db_conn, start_index=start_index, end_index=end_index) except Exception as e: logger.error(f"処理中に予期しないエラーが発生しました: {e}") import traceback traceback.print_exc() raise finally: if db_conn: db_conn.close() if __name__ == "__main__": # コマンドライン引数の設定 parser = argparse.ArgumentParser(description="腎臓移植データ処理システム - 分割実行対応版") parser.add_argument("--start", type=int, default=0, help="処理開始インデックス（0から開始、デフォルト: 0）") parser.add_argument("--limit", type=int, default=None, help="処理件数制限（未指定の場合は全件処理）") parser.add_argument("--no-clear", action="store_true", help="テーブルクリアをスキップ（--startが0以外の場合は自動でスキップ）") parser.add_argument("--optimize", choices=['O', 'OO'], help="Python最適化レベル（-O または -OO相当）") args = parser.parse_args() # Python最適化フラグの適用 if args.optimize: import sys if args.optimize == 'O': sys.dont_write_bytecode = True # -O相当 #logger.info("🚀 Python最適化レベル-O: __debug__無効化、assert文削除") elif args.optimize == 'OO': sys.dont_write_bytecode = True # -OO相当 #logger.info("🚀 Python最適化レベル-OO: __debug__無効化、assert文・docstring削除") # 最適化の効果をログ出力 #logger.info(f"📈 最適化設定: __debug__ = {__debug__}") #logger.info(f"📈 バイトコード出力: {not sys.dont_write_bytecode}") # --startが0以外の場合は自動でテーブルクリアをスキップ clear_tables = not args.no_clear and args.start == 0 if args.start != 0: logger.info(f"🔄 分割実行モード: --start={args.start} のためテーブルクリアをスキップします") main(start_index=args.start, limit=args.limit, clear_tables=clear_tables)