""" 統合版変換処理システム（コード変換表対応） """ import json import pandas as pd import numpy as np from typing import Dict, Any, List, Optional, Union import logging from datetime import datetime import re import os import csv import threading from datetime import datetime logger = logging.getLogger(__name__) class ConversionProcessor: """統合版変換処理クラス（コード変換表対応）""" # クラス変数として共有のCSVファイル情報を保持 _shared_csv_filename = None _shared_csv_initialized = False _csv_lock = threading.Lock() # スレッドセーフのためのロック（必要に応じて） def __init__(self, conversion_rules_file: str, code_mapping_file: str = None): """ 初期化 Args: conversion_rules_file: 変換ルールJSONファイルのパス code_mapping_file: コード変換表Excelファイルのパス（オプション） """ self.conversion_rules = self._load_conversion_rules(conversion_rules_file) self.code_mapping = self._load_code_mapping(code_mapping_file) if code_mapping_file else {} # 変換ログ記録用のリスト（インスタンスごと） self.conversion_logs = [] # CSVファイルのヘッダー定義 self.csv_headers = [ "処理日時", "YearNo", "テーブル名", "CSVカラム名", "対象DBカラム名", "元の値", "変換後の値", "変換タイプ", "変換詳細", "値変更フラグ" ] # ログ出力ディレクトリの作成 self.log_dir = "./logs/conversion" os.makedirs(self.log_dir, exist_ok=True) # 共有CSVファイルの初期化（最初のインスタンスのみ） if not ConversionProcessor._shared_csv_initialized: self._init_shared_csv_file() # 【NEW】複数ファイルデータを保持するための辞書 self.loaded_data = {} self._auto_load_csv_files() # ← このコメントアウトを解除 # ======================================== # 【追加2】新しいメソッド - ファイルデータ読み込み # ======================================== def load_file_data(self, file_type: str, data: pd.DataFrame): """ ファイルデータを読み込み Args: file_type: ファイルタイプ（"transplants", "d-followups", "r-followups"） data: 読み込むDataFrame """ self.loaded_data[file_type] = data # (f"📁 {file_type}データを読み込みました: {len(data)}件") def _auto_load_csv_files(self): """CSVファイルを自動読み込み（jin.pyからの読み込みを優先）""" # 既にデータが読み込まれている場合はスキップ if self.loaded_data: #logger.info("📁 データは既に読み込まれています（jin.pyから）") return try: csv_paths = { "transplants": "/csv/kan/transplants.csv", "d-followups": "/csv/kan/d-followups.csv", "r-followups": "/csv/kan/r-followups.csv" } #logger.info("📁 ConversionProcessor: CSVファイル自動読み込み開始") for file_type, path in csv_paths.items(): try: if os.path.exists(path): #logger.info(f"📄 読み込み中: {file_type} <- {path}") df = pd.read_csv(path, encoding='iso-8859-1', low_memory=False) self.load_file_data(file_type, df) #logger.info(f"✅ {file_type}: {len(df)}件のデータを読み込み完了") else: logger.warning(f"⚠️ CSVファイルが見つかりません: {path}") except Exception as e: logger.error(f"❌ {file_type}の読み込みエラー: {e}") continue #logger.info("✅ ConversionProcessor: CSV自動読み込み完了") except Exception as e: logger.error(f"❌ CSV自動読み込み処理でエラー: {e}") # ======================================== # 【追加3】新しいメソッド - クロスリファレンス処理 #### [AIへ] 回答: この関数は複数CSVファイル間の関連データを取得・マッチングする重要な処理です。全行にコメントを追加します。 # ======================================== def process_cross_reference_with_condition(self, row_data: Dict[str, Any], rule_config: Dict[str, Any]) -> Dict[str, Any]: """ 複数ファイル参照＋条件処理（改良版） 移植データ、ドナー追跡データ、レシピエント追跡データ間の関連付けを行う """ result = {} # 戻り値となる結果データを格納する辞書を初期化 # ルール設定から各ソースを取得 primary_source = rule_config.get("primary_source", {}) # 主要データソース（通常はtransplants.csv）の設定を取得 secondary_source = rule_config.get("secondary_source", {}) # 二次データソース（followupデータ等）の設定を取得 alt_secondary_source = rule_config.get("alternative_secondary_source", {}) # 代替二次データソースの設定を取得 # 主要ソース（transplants）から値を取得 primary_value = None # 主要ソースから取得する値を初期化 if primary_source.get("file_type") == "transplants": # 主要ソースがtransplantsファイルの場合の処理 primary_col = primary_source.get("source_col") # 主要ソースのカラム名を取得 primary_value = row_data.get(primary_col) # 主要ソースから実際の値を取得 # セカンダリソース（d-followups または r-followups）から値を取得 secondary_value = None # セカンダリソースからの値を初期化 secondary_file_type = secondary_source.get("file_type") # セカンダリファイルのタイプを取得 if secondary_file_type in self.loaded_data: # セカンダリファイルがメモリに読み込まれているかチェック secondary_data = self.loaded_data[secondary_file_type] # メモリからセカンダリデータを取得 # secondary_sourceのカラムを取得してtarget_columnsとして渡す secondary_col = secondary_source.get("source_col") target_cols = [secondary_col] if secondary_col else None matching_record = self._find_matching_record(row_data, secondary_data, target_cols) # 主要データとマッチするレコードを検索 if matching_record: # マッチするレコードが見つかった場合 secondary_col = secondary_source.get("source_col") # セカンダリソースのカラム名を取得 secondary_value = matching_record.get(secondary_col) # マッチしたレコードから値を取得 #else: # マッチするレコードが見つからなかった場合 # print("❌ セカンダリデータでマッチングレコードが見つかりませんでした") # 失敗メッセージをデバッグ出力 # 代替ソースも試行 ## 片岡：一旦コメントアウト #if secondary_value is None and alt_secondary_source: # セカンダリソースで値が取得できず、代替ソースが存在する場合 # print("代替ソースを試行中...") # 代替ソース試行のデバッグ出力 # alt_file_type = alt_secondary_source.get("file_type") # 代替ソースのファイルタイプを取得 # if alt_file_type in self.loaded_data: # 代替ソースがメモリに読み込まれているかチェック # alt_data = self.loaded_data[alt_file_type] # メモリから代替データを取得 # # # alternative_secondary_sourceのカラムを取得してtarget_columnsとして渡す # alt_col = alt_secondary_source.get("source_col") # alt_target_cols = [alt_col] if alt_col else None # print(f"🔍 代替ソースのデータ存在チェック対象カラム: {alt_target_cols}") # # matching_record = self._find_matching_record(row_data, alt_data, alt_target_cols) # 代替データでマッチするレコードを検索 # if matching_record: # 代替データでマッチするレコードが見つかった場合 # alt_col = alt_secondary_source.get("source_col") # 代替ソースのカラム名を取得 # secondary_value = matching_record.get(alt_col) # 代替データから値を取得 # print(f"✅ 代替ソースから取得: {alt_col} = {secondary_value}") # 代替データからの値取得成功をデバッグ出力 # secondary_valueがNoneの場合のデフォルト処理 if secondary_value is None: # セカンダリソースから値が取得できなかった場合の処理 secondary_value = "不明" # デフォルト値として"不明"を設定 # クロスリファレンスルールの適用 cross_rules = rule_config.get("cross_reference_rules", []) # 変換ルール設定からクロスリファレンスルールのリストを取得 target_col = rule_config.get("target_col") # 変換結果を格納するターゲットカラム名を取得 # ルールマッチング matched = False # ルールマッチングの成功フラグを初期化 for rule in cross_rules: # 各クロスリファレンスルールをループで処理 if (str(primary_value) == str(rule.get("primary")) and # 主要値とルールの主要値が一致し、 str(secondary_value) == str(rule.get("secondary"))): # セカンダリ値とルールのセカンダリ値が一致する場合 result[target_col] = rule.get("result") # ルールの結果値を結果辞書に設定 matched = True # マッチ成功フラグをTrueに設定 break # マッチしたのでループを終了 if not matched: # どのルールともマッチしなかった場合のフォールバック処理 # フォールバック処理 if str(primary_value) == "有": # 主要値が"有"の場合 result[target_col] = "2" # 有+不明のコード"2"を設定 else: # その他の値の場合 result[target_col] = "4" # その他+不明のコード"4"を設定 # 追加フィールドの処理 ## 片岡、一旦コメントアウトしている。 #additional_fields = rule_config.get("additional_fields", {}) # 追加フィールドの設定を取得 #for field_key, field_config in additional_fields.items(): # 各追加フィールドをループで処理 # field_result = self._process_additional_field(row_data, field_config) # 追加フィールドの処理を実行 # if field_result: # 追加フィールドの処理結果がある場合 # result.update(field_result) # 結果辞書に追加フィールドの結果をマージ # print(f"追加フィールド: {field_result}") # 追加フィールドの結果をデバッグ出力 # ログ記録 disease_name = rule_config.get("disease_name", "未指定") for field_name, field_value in result.items(): self._log_conversion( row_data=row_data, source_col=f"{primary_source.get('source_col')}+{secondary_source.get('source_col')}", target_col=field_name, original_value=f"{primary_value}+{secondary_value}", converted_value=field_value, conversion_type="cross_reference_with_condition", conversion_detail=f"疾患名: {disease_name}, クロスリファレンス変換" ) return result def _log_conversion(self, row_data: Dict[str, Any], source_col: str, target_col: str, original_value: Any, converted_value: Any, conversion_type: str, conversion_detail: str = "") -> None: """ 変換ログを記録する """ try: year_no = row_data.get('YearNo', 'N/A') log_entry = { 'timestamp': datetime.now().isoformat(), 'table_name': self.current_table_name, 'year_no': year_no, 'source_col': source_col, 'target_col': target_col, 'original_value': str(original_value) if original_value is not None else 'None', 'converted_value': str(converted_value) if converted_value is not None else 'None', 'conversion_type': conversion_type, 'conversion_detail': conversion_detail } # コンソールにログ出力 logger.debug(f"🔄 変換ログ: {target_col} = {original_value} → {converted_value} (タイプ: {conversion_type})") except Exception as e: logger.warning(f"ログ記録エラー: {e}") # ======================================== # 【追加4】新しいメソッド - レコードマッチング # ======================================== def _find_matching_record(self, transplant_row: Dict[str, Any], follow_up_data: pd.DataFrame, target_columns: List[str] = None) -> Optional[Dict[str, Any]]: """ 移植データとフォローアップデータの対応レコードを検索 YearNoで検索し、データが存在する最新レコードを優先的に取得 Args: transplant_row: 移植データの行 follow_up_data: フォローアップデータ target_columns: データ存在チェック対象のカラムリスト（Noneの場合はデフォルトの疾患カラムを使用） """ try: # キーマッピング year_no_key = transplant_row.get('YearNo') facility_key = transplant_row.get('施設名') if not year_no_key: return None # YearNoカラムを特定 year_no_columns = [col for col in follow_up_data.columns if 'YearNo' in col] if not year_no_columns: #print("❌ YeaしねrNoカラムが見つかりません") return None year_no_col = year_no_columns[0] # YearNoで完全一致検索（文字列として） year_no_matches = follow_up_data[ follow_up_data[year_no_col].astype(str) == str(year_no_key) ] if year_no_matches.empty: #print(f"❌ YearNo {year_no_key} に一致するレコードが見つかりません") return None if len(year_no_matches) == 1: #print("✅ 単一レコード発見") return year_no_matches.iloc[0].to_dict() # 🚨 修正: 複数レコードがある場合、データが存在するレコードを優先 #print(f"複数レコード存在（{len(year_no_matches)}件）、データ優先で選択中...") # まずデータが存在するレコードを探す # target_columnsが指定されていない場合はデフォルトの疾患カラムを使用 if target_columns is None or not target_columns: target_columns = ['ドナー所見:疾患:高血圧', 'ドナー所見:疾患:糖尿病', 'ドナー所見:疾患:高脂血症', 'ドナー所見:疾患:脳血管障害', 'ドナー所見:疾患:心疾患', 'ドナー所見:疾患:肝疾患'] #print(f"データ存在チェック対象カラム: {target_columns}") #print(f"チェック対象レコード数: {len(year_no_matches)}") # データが存在するレコードをフィルタリング records_with_data = [] # レコード数が多すぎる場合の保護 if len(year_no_matches) > 100: #print(f"⚠️ レコード数が多すぎます({len(year_no_matches)}件)。最初の100件のみ処理します。") year_no_matches = year_no_matches.head(100) for idx, row in year_no_matches.iterrows(): has_data = False # カラム数が多すぎる場合の保護 check_columns = target_columns[:10] if len(target_columns) > 10 else target_columns for col in check_columns: if col and col in row.index: # colがNoneでないことを確認 value = row[col] if pd.notna(value) and str(value).strip() != "": has_data = True break if has_data: records_with_data.append((idx, row)) # データが存在するレコードがある場合はそれを使用 if records_with_data: year_no_matches = pd.DataFrame([row for idx, row in records_with_data]) # 日付カラムでソート（最新優先） date_columns = [col for col in year_no_matches.columns if '調査日' in col or '日付' in col or 'DATE' in col.upper()] if date_columns: date_col = date_columns[0] #print(f"日付カラム '{date_col}' でソート中...") # 日付でソート（降順 = 新しい順） year_no_matches_sorted = year_no_matches.copy() year_no_matches_sorted[date_col] = pd.to_datetime(year_no_matches_sorted[date_col], errors='coerce') year_no_matches_sorted = year_no_matches_sorted.sort_values(date_col, ascending=False) #print("日付カラムでソート済み") selected_record = year_no_matches_sorted.iloc[0].to_dict() else: # 追跡調査の種類でソート（1年後、2年後...の順） if 'ドナー追跡調査:追跡調査の種類' in year_no_matches.columns: #print("追跡調査の種類でソート中...") def parse_follow_up_priority(val): if pd.isna(val) or str(val).strip() == "": return 999 # 空は最低優先度 val_str = str(val).strip() if "ヶ月後" in val_str: return 0 # 3ヶ月後は最高優先度 elif "年後" in val_str: try: year = int(val_str.replace("年後", "").strip()) return year # 1年後=1, 2年後=2... except: return 999 return 999 year_no_matches_sorted = year_no_matches.copy() year_no_matches_sorted['priority'] = year_no_matches_sorted['ドナー追跡調査:追跡調査の種類'].apply(parse_follow_up_priority) year_no_matches_sorted = year_no_matches_sorted.sort_values('priority', ascending=True) #print("追跡調査の種類でソート済み") selected_record = year_no_matches_sorted.iloc[0].to_dict() else: # デフォルト: 最初のレコード selected_record = year_no_matches.iloc[0].to_dict() #print("✅ 最終選択レコード決定") # 重要なフィールドの値をログ出力 important_fields = ['ドナー追跡調査:YearNo', 'ドナー追跡調査:追跡調査の種類', 'ドナー所見:疾患:糖尿病', 'ドナー所見:疾患:高血圧', 'ドナー所見:疾患:疾患名', 'ドナー所見:疾患:その他'] #print("重要フィールドの値:") for field in important_fields: if field in selected_record: value = selected_record[field] #print(f" {field}: {value}") #### [AIへ] 回答: ループで取得したデータのログ出力を追加します # ループで取得したデータのログ出力 #logger.info(f"🔍 ループ処理結果: 選択されたレコード={selected_record}") #logger.info(f"📊 ループ結果詳細: レコード数={len(selected_record) if isinstance(selected_record, dict) else 0}") return selected_record except Exception as e: #print(f"❌ レコード検索エラー: {e}") logger.error(f"レコード検索エラー: {e}") return None # ======================================== # 【追加5】新しいメソッド - 追加フィールド処理 # ======================================== def _process_additional_field(self, row_data: Dict[str, Any], field_config: Dict[str, Any]) -> Dict[str, Any]: """ 追加フィールドの処理（ETCやETC_CMNT等） """ result = {} target_col = field_config.get("target_col") condition = field_config.get("condition") source = field_config.get("source", {}) rules = field_config.get("rules", {}) if not target_col or not source: return result # ソースデータを取得 file_type = source.get("file_type") source_col = source.get("source_col") value = None if file_type == "transplants": # 移植データから直接取得 value = row_data.get(source_col) elif file_type in self.loaded_data: # フォローアップデータから取得 follow_up_data = self.loaded_data[file_type] # sourceのカラムを取得してtarget_columnsとして渡す # source_colは既に435行目で定義済み field_target_cols = [source_col] if source_col else None #print(f"🔍 追加フィールドのデータ存在チェック対象カラム: {field_target_cols}") matching_record = self._find_matching_record(row_data, follow_up_data, field_target_cols) if matching_record: value = matching_record.get(source_col) #print(f"追加フィールド処理: {target_col}, condition: {condition}, value: {value}") # 条件に基づいて処理 if condition == "has_other_disease_value": # 「その他」フィールドに値がある場合の処理 if value and pd.notna(value) and str(value).strip(): result[target_col] = rules.get("has_value", "3") else: result[target_col] = rules.get("default", None) elif condition == "other_disease_comment": # 疾患名コメントの処理 if value and pd.notna(value) and str(value).strip(): if rules.get("has_value") == "pass_through": result[target_col] = str(value).strip() else: result[target_col] = rules.get("has_value") else: result[target_col] = rules.get("empty_value", None) #### [AIへ] 回答: 処理結果のログ出力を追加します # 処理結果のログ出力 #logger.info(f"✅ process_cross_reference_with_condition処理完了: 結果={result}") #logger.debug(f"🔍 詳細結果: primary_value={locals().get('primary_value')}, secondary_value={locals().get('secondary_value')}") return result def _init_shared_csv_file(self): """共有CSVファイルを初期化（最初のインスタンスのみ実行）""" # セッション開始時刻 session_start = datetime.now().strftime("%Y%m%d_%H%M%S") # 共有CSVファイル名を設定 ConversionProcessor._shared_csv_filename = os.path.join( self.log_dir, f"conversion_log_{session_start}.csv" ) # CSVファイルの初期化（ヘッダー書き込み） with open(ConversionProcessor._shared_csv_filename, 'w', newline='', encoding='utf-8-sig') as csvfile: writer = csv.DictWriter(csvfile, fieldnames=self.csv_headers) writer.writeheader() ConversionProcessor._shared_csv_initialized = True #logger.info(f"📄 統合変換ログCSVファイルを作成しました: {ConversionProcessor._shared_csv_filename}") def _write_conversion_log(self, log_data: Dict[str, Any]): """変換ログを共有CSVファイルに追記""" # スレッドセーフのためのロック（必要に応じて） with open(ConversionProcessor._shared_csv_filename, 'a', newline='', encoding='utf-8-sig') as csvfile: writer = csv.DictWriter(csvfile, fieldnames=self.csv_headers) writer.writerow(log_data) @classmethod def get_shared_csv_filename(cls): """共有CSVファイル名を取得""" return cls._shared_csv_filename @classmethod def reset_shared_csv(cls): """共有CSVファイルをリセット（新しいセッション開始時に使用）""" cls._shared_csv_filename = None cls._shared_csv_initialized = False def _load_conversion_rules(self, file_path: str) -> Dict[str, Any]: """変換ルールJSONファイルを読み込む""" try: with open(file_path, 'r', encoding='utf-8') as f: return json.load(f) except FileNotFoundError: logger.error(f"変換ルールファイルが見つかりません: {file_path}") raise except json.JSONDecodeError as e: logger.error(f"変換ルールファイルの形式が不正です: {e}") raise def _load_code_mapping(self, file_path: str) -> Dict[str, Any]: """コード変換表Excelファイルを読み込む""" if not file_path or not os.path.exists(file_path): logger.warning(f"コード変換表ファイルが見つかりません: {file_path}") return {} try: # 複数シートを読み込む mapping_dict = pd.read_excel(file_path, sheet_name=None) code_mapping = {} for sheet_name, df in mapping_dict.items(): #logger.info(f"コード変換表シート読み込み: {sheet_name}") # シートの内容を辞書化 if not df.empty: # 最初の列をキー、2番目の列を値とする辞書を作成 if len(df.columns) >= 2: sheet_mapping = {} for _, row in df.iterrows(): key = str(row.iloc[0]).strip() if pd.notna(row.iloc[0]) else "" value = str(row.iloc[1]).strip() if pd.notna(row.iloc[1]) else "" if key and value: sheet_mapping[key] = value code_mapping[sheet_name] = sheet_mapping #logger.info(f"コード変換表読み込み完了: {len(code_mapping)} シート") return code_mapping except Exception as e: logger.error(f"コード変換表読み込みエラー: {e}") return {} #### [AIへ] 回答: はい、その通りです。jin.pyのprocess_transplant_data()関数から呼ばれる最初のメソッドで、データ変換ルール適用のエントリーポイントです。 def apply_conversion_rules(self, table_name: str, transplant_row: pd.Series, recipient_rows: List[tuple] = None, donor_rows: List[tuple] = None) -> Dict[str, Any]: """ 指定されたテーブルの変換ルールをすべて適用して変換結果を返す """ result = {} self.current_table_name = table_name table_rules = self.conversion_rules.get(table_name, {}) #print(f"🔍 テーブル {table_name} の変換ルール数: {len(table_rules)}") #print(f"🔍 テーブルルール: {list(table_rules.keys())}") for field_key, rule_config in table_rules.items(): if rule_config.get("target_table") != table_name: continue target_col = rule_config.get("target_col") conversion_type = rule_config.get("type") #print("$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$") #print(f"field_key: {field_key}") #print(f"target_col: {target_col}") #print(f"conversion_type: {conversion_type}") # cross_reference_with_conditionタイプの特別処理 if conversion_type == "cross_reference_with_condition": #print("🚨 cross_reference_with_condition タイプを検出! 処理開始...") #### [AIへ] 回答: はい、pandasのSeriesオブジェクトをPythonの辞書(dict)に変換しています。これはCSVデータを操作しやすくするためです。 row_data = transplant_row.to_dict() if hasattr(transplant_row, 'to_dict') else dict(transplant_row) #### [AIへ] 回答: おっしゃる通りです。この関数は他の箇所でも呼ばれており、重複処理の可能性があります。パフォーマンス改善の余地があります。 cross_result = self.process_cross_reference_with_condition(row_data, rule_config) #print(f"🎯 cross_reference結果: {cross_result}") # 複数フィールドの結果をマージ if isinstance(cross_result, dict): result.update(cross_result) #print(f"🔄 結果をマージしました: {cross_result}") continue # 次のルールへ # 複数フィールドを返すタイプの場合はtarget_colがNoneでも処理する if conversion_type in ["conditional_multi_field", "multi_checkbox_split", "multi_field_conditional"]: converted_value = self._convert_value( conversion_type, rule_config, transplant_row, recipient_rows, donor_rows ) # 複数フィールドの結果をマージ if isinstance(converted_value, dict): result.update(converted_value) elif converted_value is not None and target_col: # 単一フィールドの場合（後方互換性） result[target_col] = converted_value continue # 次のルールへ # 通常の処理（既存処理） if not target_col or not conversion_type: #print(f"⚠️ target_colまたはconversion_typeが不正: {field_key}") continue # 変換タイプに応じて処理を分岐 converted_value = self._convert_value( conversion_type, rule_config, transplant_row, recipient_rows, donor_rows ) if converted_value is not None: result[target_col] = converted_value #print(f"✅ {target_col} に値を設定: {converted_value}") target_col = rule_config.get("target_col","") source_col = rule_config.get("source_col", "") result["target_col"] = target_col result["source_col"] = source_col return result def _convert_value(self, conversion_type: str, rule_config: Dict[str, Any], transplant_row: pd.Series, recipient_rows: List[tuple] = None, donor_rows: List[tuple] = None) -> Any: """ 変換タイプに応じて値を変換する """ source_col = rule_config.get("source_col") target_col = rule_config.get("target_col") rules = rule_config.get("rules", {}) # time_conversionタイプの処理を追加 if conversion_type == "time_conversion": return self._apply_time_conversion(rule_config, transplant_row, recipient_rows, donor_rows) # conditional_textタイプの特別処理 if conversion_type == "conditional_text": return self._apply_conditional_text(rule_config, transplant_row, recipient_rows, donor_rows) # データソースから値を取得 raw_value = self._get_raw_value( conversion_type, source_col, transplant_row, recipient_rows, donor_rows ) # 値がNULLまたは空の場合でもログを記録 if raw_value is None or (isinstance(raw_value, str) and raw_value.strip() == ""): # YearNoを取得（識別用） year_no = "" if hasattr(transplant_row, 'get'): year_no = transplant_row.get('YearNo', '') # ログデータの作成（値なしの場合） log_data = { "処理日時": datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3], "YearNo": year_no, "テーブル名": getattr(self, 'current_table_name', ''), "CSVカラム名": source_col, "対象DBカラム名": target_col, "元の値": "(値なし)", "変換後の値": "(変換スキップ)", "変換タイプ": conversion_type, "変換詳細": "元データなしのためスキップ", "値変更フラグ": "変更なし" } # CSVファイルに書き込み self._write_conversion_log(log_data) # メモリにも保存 self.conversion_logs.append(log_data) # デバッグログ出力 if logger.isEnabledFor(logging.DEBUG): console_msg = ( f"[値なし] YearNo:{year_no} | " f"{self.current_table_name}.{target_col} | " f"{source_col}: 値なしのためスキップ" ) logger.debug(console_msg) return None # コード変換表を参照して変換ルールを拡張 enhanced_rules = self._enhance_rules_with_code_mapping(source_col, rules) # 変換前の値を記録 original_value = raw_value converted_value = None conversion_detail = "" # 変換タイプに応じて変換処理を実行 if conversion_type == "single_map": converted_value = self._apply_single_map(raw_value, enhanced_rules) conversion_detail = "単一値マッピング" elif conversion_type == "multi_flag": converted_value = self._apply_multi_flag(raw_value, enhanced_rules) conversion_detail = "複数フラグ変換" elif conversion_type == "multi_flag_split": converted_value = self._apply_multi_flag_split(raw_value, enhanced_rules) conversion_detail = "複数フラグ分割変換" elif conversion_type == "date_format": converted_value = self._apply_date_format(raw_value, rule_config.get("format", "%Y%m%d")) conversion_detail = f"日付フォーマット変換 (形式: {rule_config.get('format', '%Y%m%d')})" elif conversion_type == "pass_through": converted_value = raw_value conversion_detail = "変換なし（そのまま通過）" elif conversion_type == "numeric_conversion": converted_value = self._apply_numeric_conversion(raw_value, enhanced_rules) conversion_detail = "数値変換" elif conversion_type == "conditional_multi_field": converted_value = self._apply_conditional_multi_field(rule_config, transplant_row, recipient_rows, donor_rows) conversion_detail = "条件付き複数フィールド変換" elif conversion_type in ["transplant_time", "recipient_followup", "donor_followup"]: # 日付フォーマット指定がある場合は日付変換を適用 if "format" in rules: converted_value = self._apply_date_format(raw_value, rules.get("format", "%Y%m%d")) conversion_detail = f"日付フォーマット変換 (形式: {rules.get('format', '%Y%m%d')})" # defaultがpass_throughの場合はそのまま返す elif rules.get("default") == "pass_through": converted_value = raw_value conversion_detail = "変換なし（pass_through指定）" # defaultがnumeric_conversionの場合は数値変換を適用 elif rules.get("default") == "numeric_conversion": converted_value = self._apply_numeric_conversion(raw_value, enhanced_rules) conversion_detail = "数値変換（numeric_conversion指定）" # それ以外は通常のマッピング変換 else: converted_value = self._apply_single_map(raw_value, enhanced_rules) conversion_detail = "マッピング変換" elif conversion_type == "multi_checkbox_split": converted_value = self._apply_multi_checkbox_split(rule_config, transplant_row) conversion_detail = "複数チェックボックス分割" elif conversion_type == "multi_field_conditional": converted_value = self._apply_multi_field_conditional(rule_config, transplant_row) conversion_detail = "複数フィールド条件付き変換" elif conversion_type == "check_exists": converted_value = self._apply_check_exists(raw_value) conversion_detail = "存在チェック変換" elif conversion_type == "code_mapping": converted_value = self._apply_code_mapping(raw_value, source_col) conversion_detail = "コード変換表による変換" elif conversion_type == "cross_reference_with_condition": # row_dataを作成（transplant_rowから） row_data = transplant_row.to_dict() if hasattr(transplant_row, 'to_dict') else dict(transplant_row) #### [AIへ] 回答: はい、ここでもprocess_cross_reference_with_conditionが呼ばれています。これも重複処理の一例です。 return self.process_cross_reference_with_condition(row_data, rule_config) else: logger.warning(f"未対応の変換タイプ: {conversion_type}") converted_value = raw_value conversion_detail = "未対応タイプ（変換なし）" # 変換結果をログに出力（値が変更された場合、または重要な変換の場合） # 変換結果をログに出力（値が変更された場合、または重要な変換の場合） #if converted_value is not None: # YearNoを取得（識別用） year_no = "" if hasattr(transplant_row, 'get'): year_no = transplant_row.get('YearNo', '') # 値変更フラグ value_changed = str(original_value) != str(converted_value) # ログデータの作成 log_data = { "処理日時": datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3], "YearNo": year_no, "テーブル名": getattr(self, 'current_table_name', ''), "CSVカラム名": source_col, "対象DBカラム名": target_col, "元の値": str(original_value) if original_value is not None else "(値なし)", "変換後の値": str(converted_value) if converted_value is not None else "変換なし", "変換タイプ": conversion_type, "変換詳細": conversion_detail, "値変更フラグ": "変更あり" if value_changed else "変更なし" } # CSVファイルに書き込み self._write_conversion_log(log_data) # メモリにも保存 self.conversion_logs.append(log_data) # コンソール出力（簡易形式） if value_changed: console_msg = ( f"[変換] YearNo:{year_no} | " f"{self.current_table_name}.{target_col} | " f"{source_col}: '{original_value}' → '{converted_value}' | " f"{conversion_detail}" ) #logger.info(console_msg) else: # デバッグモードの場合のみ変更なしも出力 if logger.isEnabledFor(logging.DEBUG): console_msg = ( f"[変換なし] YearNo:{year_no} | " f"{self.current_table_name}.{target_col} | " f"{source_col}: '{original_value}'" ) logger.debug(console_msg) return converted_value def _apply_multi_checkbox_split(self, rule_config: Dict[str, Any], transplant_row: pd.Series) -> Dict[str, Any]: """複数チェックボックスの分割変換を適用""" source_col = rule_config.get("source_col") config = rule_config.get("config", {}) mappings = config.get("mappings", {}) result = {} raw_value = transplant_row.get(source_col) if pd.isna(raw_value): return result value_str = str(raw_value).strip() # 各マッピングをチェック for keyword, target_col in mappings.items(): if keyword in value_str: result[target_col] = 1 else: result[target_col] = 0 return result def _apply_multi_field_conditional(self, rule_config: Dict[str, Any], transplant_row: pd.Series) -> Dict[str, Any]: """複数フィールドの条件付き変換を適用""" config = rule_config.get("config", {}) source_cols = config.get("source_cols", []) rules = config.get("rules", {}) result = {} # 各ソースカラムをチェック for col in source_cols: value = transplant_row.get(col) if pd.notna(value) and str(value).strip(): value_str = str(value).strip() # ルールに一致する場合、対応するフィールドを設定 for rule_key, field_mappings in rules.items(): if rule_key in value_str: for target_col, target_value in field_mappings.items(): result[target_col] = target_value break return result def _apply_check_exists(self, raw_value: Any) -> Any: """値の存在チェック変換を適用""" if pd.isna(raw_value) or str(raw_value).strip() == "": return "0" # 無 else: return "1" # 有 def _apply_time_conversion(self, rule_config: Dict[str, Any], transplant_row: pd.Series, recipient_rows: List[tuple] = None, donor_rows: List[tuple] = None) -> Any: """ 時間変換を適用（時間と分を分単位に変換） """ source_col_hour = rule_config.get("source_col_hour") source_col_minute = rule_config.get("source_col_minute") rules = rule_config.get("rules", {}) convert_to = rules.get("convert_to", "minutes") # データソースから値を取得 conversion_type = rule_config.get("type", "transplant_time") # 時間の値を取得 hour_value = self._get_raw_value( conversion_type, source_col_hour, transplant_row, recipient_rows, donor_rows ) # 分の値を取得 minute_value = self._get_raw_value( conversion_type, source_col_minute, transplant_row, recipient_rows, donor_rows ) # 数値変換 try: # 時間を数値に変換（存在しない場合は0） hours = 0 if pd.notna(hour_value) and str(hour_value).strip() != "": hours = float(str(hour_value).strip()) # 分を数値に変換（存在しない場合は0） minutes = 0 if pd.notna(minute_value) and str(minute_value).strip() != "": minutes = float(str(minute_value).strip()) # 両方とも0の場合はNoneを返す if hours == 0 and minutes == 0: return None # 分単位に変換 if convert_to == "minutes": total_minutes = int(hours * 60 + minutes) # tinyint(3) unsignedの範囲チェック（0-255） #if total_minutes > 255: # logger.warning(f"時間変換結果が範囲外: {total_minutes}分 (最大255分)") # return 255 # 上限値で制限 if total_minutes < 0: logger.warning(f"時間変換結果が負の値: {total_minutes}分") return 0 # 下限値で制限 return total_minutes else: logger.warning(f"未対応の時間変換タイプ: {convert_to}") return None except (ValueError, TypeError) as e: logger.warning(f"時間変換エラー: 時間={hour_value}, 分={minute_value}, エラー={e}") return None def _apply_code_mapping(self, raw_value: Any, source_col: str) -> Any: """コード変換表を使用した変換を適用""" if pd.isna(raw_value): return None value_str = str(raw_value).strip() # コード変換表から該当するマッピングを検索 for sheet_name, sheet_mapping in self.code_mapping.items(): if source_col in sheet_name: if value_str in sheet_mapping: return sheet_mapping[value_str] return value_str # マッピングが見つからない場合は元の値を返す def _apply_conditional_text(self, rule_config: Dict[str, Any], transplant_row: pd.Series, recipient_rows: List[tuple] = None, donor_rows: List[tuple] = None) -> Any: """条件付きテキスト変換を適用""" source_cols = rule_config.get("source_cols", []) rules = rule_config.get("rules", {}) combine_method = rules.get("combine_method", "first_non_empty") # 各ソースカラムから値を取得 values = [] for col in source_cols: value = transplant_row.get(col) if col in transplant_row.index else None if pd.notna(value) and str(value).strip(): values.append(str(value).strip()) # 結合方法に応じて処理 if combine_method == "first_non_empty": # 最初の空でない値を返す return values[0] if values else None elif combine_method == "concatenate": # 全ての値を結合 separator = rules.get("separator", " ") return separator.join(values) if values else None else: return None def _enhance_rules_with_code_mapping(self, source_col: str, base_rules: Dict[str, Any]) -> Dict[str, Any]: """ コード変換表を参照して変換ルールを拡張する Args: source_col: ソースカラム名 base_rules: 基本変換ルール Returns: 拡張された変換ルール """ enhanced_rules = base_rules.copy() # コード変換表を検索 for sheet_name, sheet_mapping in self.code_mapping.items(): # ソースカラム名とシート名またはマッピングキーが一致する場合 if source_col in sheet_name or any(source_col in key for key in sheet_mapping.keys()): # マッピングルールを追加 for key, value in sheet_mapping.items(): if key not in enhanced_rules: # 数値変換を試行 try: if value.isdigit(): enhanced_rules[key] = int(value) else: enhanced_rules[key] = value except: enhanced_rules[key] = value return enhanced_rules def _get_raw_value(self, conversion_type: str, source_col: str, transplant_row: pd.Series, recipient_rows: List[tuple] = None, donor_rows: List[tuple] = None) -> Any: """データソースから生の値を取得する""" if conversion_type == "transplant_time": # 移植時データから取得 return transplant_row.get(source_col) if source_col in transplant_row.index else None elif conversion_type == "recipient_followup" or conversion_type == "conditional_multi_field": print(f"🔍 ここにきてない@@@@@@@@@@@@@@@@@@@@@@@@@@@@ recipient_followup row_data: {source_col}") # レシピエントフォローアップデータから取得（値のある最新データを優先） if recipient_rows: for _, row_data in recipient_rows: print(f"🔍 @@@@@@@@@@@@@@@@@@@@@@@@@@@@ recipient_followup row_data: {source_col}") if source_col in row_data.index: value = row_data.get(source_col) print(f"🔍 @@@@@@@@@@@@@@@@@@@@@@@@@@@@ recipient_followup row_data: {source_col} {value}") # 値が存在し、かつ空文字でない場合に返す if pd.notna(value) and str(value).strip() != "": return value print(f"🔍 @@@@@@@@@@@@@@@@@@@@@@@@@@@@ recipient_followup row_data: {source_col} データなし") return None elif conversion_type == "donor_followup": # ドナーフォローアップデータから取得（値のある最新データを優先） if donor_rows: for _, row_data in donor_rows: if source_col in row_data.index: value = row_data.get(source_col) # 値が存在し、かつ空文字でない場合に返す if pd.notna(value) and str(value).strip() != "": return value return None else: # デフォルトは移植時データから取得 return transplant_row.get(source_col) if source_col in transplant_row.index else None def _apply_single_map(self, raw_value: Any, rules: Dict[str, Any]) -> Any: """単一値マッピング変換を適用""" # nullの場合もdefaultルールを適用する if pd.isna(raw_value): if "default" in rules: default_value = rules["default"] if default_value == "pass_through": return raw_value # nullをそのまま返す elif default_value == "numeric_conversion": return None # nullの数値変換は不可能なのでNone return default_value # "4"などの値を返す else: return None value_str = str(raw_value).strip() # 空文字の場合の処理 if value_str == "": if "" in rules: return rules[""] elif "default" in rules: default_value = rules["default"] if default_value == "pass_through": return raw_value elif default_value == "numeric_conversion": return None return default_value else: return None # 直接マッピング if value_str in rules: return rules[value_str] # 部分一致検索 for key, mapped_value in rules.items(): if key != "default" and (key in value_str or value_str in key): return mapped_value # デフォルト値 if "default" in rules: default_value = rules["default"] if default_value == "pass_through": return raw_value elif default_value == "numeric_conversion": return self._apply_numeric_conversion(raw_value, {}) return default_value return None def _apply_multi_flag(self, raw_value: Any, rules: Dict[str, Any]) -> Dict[str, Any]: """複数フラグ変換を適用""" result = {} if pd.isna(raw_value): return result value_str = str(raw_value).strip() # 各ルールに対してチェック for key, flag_col in rules.items(): if key != "default": if key in value_str: result[flag_col] = 1 else: result[flag_col] = 0 return result def _apply_multi_flag_split(self, raw_value: Any, rules: Dict[str, Any]) -> Dict[str, Any]: """複数フラグ分割変換を適用""" result = {} if pd.isna(raw_value): return result value_str = str(raw_value).strip() # カンマやセミコロンで分割 values = re.split(r'[,;、]', value_str) # 各ルールに対してチェック for key, flag_col in rules.items(): if key != "default": result[flag_col] = 0 for val in values: if key in val.strip(): result[flag_col] = 1 break return result def _apply_numeric_conversion(self, raw_value: Any, rules: Dict[str, Any]) -> Any: """数値変換を適用""" if pd.isna(raw_value): return None value_str = str(raw_value).strip() # 空文字の場合 if value_str == "": return None # 未測定系の値の場合 if value_str in ["未測定", "不明", "検査せず", "未実施", "ND", "N/A", "NA"]: return None # 数値以外の文字を除去して数値変換を試行 try: # カンマや全角数字を処理 value_str = value_str.replace(",", "").replace("，", "") value_str = value_str.translate(str.maketrans("０１２３４５６７８９", "0123456789")) # 数値部分を抽出 numeric_match = re.search(r'[-+]?[0-9]*\.?[0-9]+([eE][-+]?[0-9]+)?', value_str) if numeric_match: numeric_value = float(numeric_match.group()) # 小数点以下の桁数を調整 if numeric_value == int(numeric_value): return int(numeric_value) else: return round(numeric_value, 3) else: logger.warning(f"数値変換に失敗: {raw_value}") return None except (ValueError, TypeError) as e: logger.warning(f"数値変換エラー: {raw_value} - {e}") return None def _apply_date_format(self, raw_value: Any, format_str: str) -> str: """日付フォーマット変換を適用（年月対応版）""" if pd.isna(raw_value): return None value_str = str(raw_value).strip() # 既に適切なフォーマットの場合はそのまま返す if format_str == "%Y%m%d" and re.match(r'^\d{8}$', value_str): return value_str elif format_str == "%Y%m" and re.match(r'^\d{6}$', value_str): return value_str # 様々な日付フォーマットを試行（年月日） date_formats_ymd = [ "%Y/%m/%d", "%Y-%m-%d", "%Y年%m月%d日", "%m/%d/%Y", "%d/%m/%Y", "%Y.%m.%d", "%Y/%m/%d %H:%M:%S", # 時刻付き "%Y-%m-%d %H:%M:%S" # 時刻付き ] # 年月のみのフォーマットを試行 date_formats_ym = [ "%Y/%m", "%Y-%m", "%Y年%m月", "%Y.%m" ] # 出力フォーマットが年月日の場合 if format_str == "%Y%m%d": # まず年月日フォーマットを試行 for fmt in date_formats_ymd: try: dt = datetime.strptime(value_str, fmt) return dt.strftime(format_str) except ValueError: continue # 年月のみの場合は1日を追加して変換 for fmt in date_formats_ym: try: dt = datetime.strptime(value_str, fmt) # 1日を追加 dt = dt.replace(day=1) return dt.strftime(format_str) except ValueError: continue # 出力フォーマットが年月の場合 elif format_str == "%Y%m": # まず年月フォーマットを試行 for fmt in date_formats_ym: try: dt = datetime.strptime(value_str, fmt) return dt.strftime(format_str) except ValueError: continue # 年月日フォーマットからも年月を抽出 for fmt in date_formats_ymd: try: dt = datetime.strptime(value_str, fmt) return dt.strftime(format_str) except ValueError: continue # その他のフォーマット（既存の処理） else: for fmt in date_formats_ymd + date_formats_ym: try: dt = datetime.strptime(value_str, fmt) return dt.strftime(format_str) except ValueError: continue # 数値の場合（Excel日付シリアル値など） try: if value_str.isdigit(): if len(value_str) == 8: return value_str elif len(value_str) == 6: if format_str == "%Y%m": return value_str else: return value_str + "01" # 1日を追加 elif len(value_str) == 4: # 年のみ if format_str == "%Y%m": return value_str + "01" # 1月を追加 else: return value_str + "0101" # 1月1日を追加 except: pass logger.warning(f"日付変換に失敗: {raw_value} (フォーマット: {format_str})") return None def _apply_conditional_multi_field(self, rule_config: Dict[str, Any], transplant_row: pd.Series, recipient_rows: List[tuple] = None, donor_rows: List[tuple] = None) -> Dict[str, Any]: """条件付き複数フィールド変換を適用""" source_col = rule_config.get("source_col") config = rule_config.get("config", {}) result = {} # データソースから値を取得 conversion_type = "recipient_followup" if rule_config.get("type") == "conditional_multi_field" else rule_config.get("type", "transplant_time") raw_value = self._get_raw_value( conversion_type, source_col, transplant_row, recipient_rows, donor_rows ) # 値が存在しない場合は何もしない if pd.isna(raw_value) or str(raw_value).strip() == "": return result value_str = str(raw_value).strip() # value_based_mappings が存在する場合（新機能） if "value_based_mappings" in config: value_mappings = config["value_based_mappings"] # 該当する値のマッピングを検索 for value_pattern, field_mappings in value_mappings.items(): if value_pattern == value_str or (value_pattern == "default" and value_str not in [k for k in value_mappings.keys() if k != "default"]): for target_col, target_value in field_mappings.items(): if isinstance(target_value, dict): # ネストした条件処理 if target_value.get("default") == "pass_through": result[target_col] = raw_value else: mapped_val = self._apply_single_map(raw_value, target_value) if mapped_val is not None: result[target_col] = mapped_val else: result[target_col] = target_value break # 従来の条件ベース処理（後方互換性のため） elif "condition" in config: condition = config.get("condition", "has_value") target_fields = config.get("target_fields", {}) condition_met = False if condition == "has_value": condition_met = True # 既に値の存在は確認済み elif condition == "is_empty": condition_met = False # 値が存在するのでFalse elif condition == "equals": target_value = config.get("condition_value") condition_met = (value_str == str(target_value)) elif condition == "contains": target_value = config.get("condition_value") condition_met = (str(target_value) in value_str) if condition_met: for target_col, field_config in target_fields.items(): if isinstance(field_config, str): result[target_col] = field_config elif isinstance(field_config, dict): if field_config.get("default") == "pass_through": result[target_col] = raw_value else: mapped_value = self._apply_single_map(raw_value, field_config) if mapped_value is not None: result[target_col] = mapped_value elif isinstance(field_config, (int, float)): result[target_col] = field_config return result def get_table_rules(self, table_name: str) -> Dict[str, Any]: """指定されたテーブルの変換ルールを取得""" return self.conversion_rules.get(table_name, {}) def validate_conversion_rules(self, table_name: str) -> List[str]: """変換ルールの検証""" errors = [] table_rules = self.get_table_rules(table_name) for field_key, rule_config in table_rules.items(): if "target_table" not in rule_config: errors.append(f"{field_key}: target_table が設定されていません") if "type" not in rule_config: errors.append(f"{field_key}: type が設定されていません") if "target_col" not in rule_config: errors.append(f"{field_key}: target_col が設定されていません") conversion_type = rule_config.get("type") if conversion_type not in ["single_map", "multi_flag", "multi_flag_split", "date_format", "pass_through", "numeric_conversion", "transplant_time", "recipient_followup", "donor_followup", "conditional_text", "cross_reference_with_condition"]: errors.append(f"{field_key}: 未対応の変換タイプ {conversion_type}") return errors def get_code_mapping_stats(self) -> Dict[str, Any]: """コード変換表の統計情報を取得""" stats = { "total_sheets": len(self.code_mapping), "sheet_details": {} } for sheet_name, sheet_mapping in self.code_mapping.items(): stats["sheet_details"][sheet_name] = { "total_mappings": len(sheet_mapping), "sample_mappings": dict(list(sheet_mapping.items())[:5]) # 最初の5件を表示 } return stats def _init_csv_file(self): """CSVファイルを初期化（ヘッダーを書き込む）""" with open(self.csv_filename, 'w', newline='', encoding='utf-8-sig') as csvfile: writer = csv.DictWriter(csvfile, fieldnames=self.csv_headers) writer.writeheader() #logger.info(f"📄 変換ログCSVファイルを作成しました: {self.csv_filename}") def _write_conversion_log(self, log_data: Dict[str, Any]): """変換ログを共有CSVファイルに追記""" if not ConversionProcessor._shared_csv_filename: logger.error("共有CSVファイルが初期化されていません") return try: with open(ConversionProcessor._shared_csv_filename, 'a', newline='', encoding='utf-8-sig') as csvfile: writer = csv.DictWriter(csvfile, fieldnames=self.csv_headers) writer.writerow(log_data) csvfile.flush() # 強制的にディスクに書き込む os.fsync(csvfile.fileno()) # OSレベルで確実に書き込む except Exception as e: logger.error(f"CSV書き込みエラー: {e}") import traceback traceback.print_exc() def get_conversion_summary(self) -> Dict[str, Any]: """変換ログのサマリーを取得""" if not self.conversion_logs: return {} df = pd.DataFrame(self.conversion_logs) # サマリー統計 summary = { "総変換数": len(df), "値変更あり": len(df[df["値変更フラグ"] == "変更あり"]), "値変更なし": len(df[df["値変更フラグ"] == "変更なし"]), "テーブル別件数": df.groupby("テーブル名").size().to_dict(), "変換タイプ別件数": df.groupby("変換タイプ").size().to_dict() } return summary def save_conversion_summary(self): """変換サマリーをファイルに保存""" summary = self.get_conversion_summary() if not summary: return summary_file = os.path.join(self.log_dir, f"conversion_summary_{self.session_start}.txt") with open(summary_file, 'w', encoding='utf-8') as f: f.write("=== 変換処理サマリー ===\n") f.write(f"処理日時: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n") f.write(f"CSVログファイル: {self.csv_filename}\n\n") f.write(f"総変換数: {summary.get('総変換数', 0)}\n") f.write(f"値変更あり: {summary.get('値変更あり', 0)}\n") f.write(f"値変更なし: {summary.get('値変更なし', 0)}\n\n") f.write("【テーブル別件数】\n") for table, count in summary.get('テーブル別件数', {}).items(): f.write(f" {table}: {count}件\n") f.write("\n【変換タイプ別件数】\n") for conv_type, count in summary.get('変換タイプ別件数', {}).items(): f.write(f" {conv_type}: {count}件\n") #logger.info(f"📊 変換サマリーを保存しました: {summary_file}") @classmethod def save_global_conversion_summary(cls, conversion_processors: Dict[str, 'ConversionProcessor']): """全ConversionProcessorの変換サマリーをファイルに保存""" # 全インスタンスのログを結合 all_logs = [] for name, processor in conversion_processors.items(): all_logs.extend(processor.conversion_logs) # 変換ログが空でも必ずサマリーファイルを作成 if not all_logs: # 空のサマリーを作成 summary = { "総変換数": 0, "値変更あり": 0, "値変更なし": 0, "テーブル別件数": {}, "変換タイプ別件数": {}, "プロセッサ別件数": {name: len(proc.conversion_logs) for name, proc in conversion_processors.items()} } # サマリーファイル名（CSVファイル名から生成） if cls._shared_csv_filename: base_name = os.path.splitext(os.path.basename(cls._shared_csv_filename))[0] summary_file = os.path.join(os.path.dirname(cls._shared_csv_filename), f"{base_name}_summary.txt") else: summary_file = "./logs/conversion/conversion_summary.txt" with open(summary_file, 'w', encoding='utf-8') as f: f.write("=== 変換処理サマリー（ログなし） ===\n") f.write(f"実行日時: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n") for key, value in summary.items(): f.write(f"{key}: {value}\n") f.write("\n注意: この実行では変換ログが記録されませんでした。\n") f.write("データ変換が実行されなかった、またはエラーで中断された可能性があります。\n") return df = pd.DataFrame(all_logs) # サマリー統計 summary = { "総変換数": len(df), "値変更あり": len(df[df["値変更フラグ"] == "変更あり"]), "値変更なし": len(df[df["値変更フラグ"] == "変更なし"]), "テーブル別件数": df.groupby("テーブル名").size().to_dict(), "変換タイプ別件数": df.groupby("変換タイプ").size().to_dict(), "プロセッサ別件数": {name: len(proc.conversion_logs) for name, proc in conversion_processors.items()} } # サマリーファイル名（CSVファイル名から生成） if cls._shared_csv_filename: base_name = os.path.splitext(os.path.basename(cls._shared_csv_filename))[0] summary_file = os.path.join(os.path.dirname(cls._shared_csv_filename), f"{base_name}_summary.txt") else: summary_file = "./logs/conversion/conversion_summary.txt" with open(summary_file, 'w', encoding='utf-8') as f: f.write("=== 変換処理サマリー ===\n") f.write(f"処理日時: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n") f.write(f"CSVログファイル: {cls._shared_csv_filename}\n\n") f.write(f"総変換数: {summary.get('総変換数', 0)}\n") f.write(f"値変更あり: {summary.get('値変更あり', 0)}\n") f.write(f"値変更なし: {summary.get('値変更なし', 0)}\n\n") f.write("【プロセッサ別件数】\n") for name, count in summary.get('プロセッサ別件数', {}).items(): f.write(f" {name}: {count}件\n") f.write("\n【テーブル別件数】\n") for table, count in summary.get('テーブル別件数', {}).items(): f.write(f" {table}: {count}件\n") f.write("\n【変換タイプ別件数】\n") for conv_type, count in summary.get('変換タイプ別件数', {}).items(): f.write(f" {conv_type}: {count}件\n") #logger.info(f"📊 統合変換サマリーを保存しました: {summary_file}")