""" エラーレポート生成モジュール データベースのテーブル定義を取得し、エラー情報と組み合わせて 詳細なエラーレポートを生成する """ import pandas as pd import pymysql from typing import Dict, List, Any, Optional import logging import re from datetime import datetime import os logger = logging.getLogger(__name__) class ErrorReportGenerator: """エラーレポート生成クラス""" def __init__(self, db_connection: pymysql.connections.Connection): """ 初期化 Args: db_connection: データベース接続 """ self.connection = db_connection self.table_definitions = {} self.error_analysis_rules = self._init_error_analysis_rules() def _init_error_analysis_rules(self) -> Dict[str, Dict[str, str]]: """エラー分析ルールを初期化""" return { "Data too long": { "pattern": r"Data too long for column '(.+)' at row", "対処方法": "カラムのサイズを拡張するか、データを切り詰める", "エラータイプ": "データ長超過" }, "Duplicate entry": { "pattern": r"Duplicate entry '(.+)' for key", "対処方法": "重複チェックを追加するか、既存データを更新する", "エラータイプ": "重複エラー" }, "Incorrect integer value": { "pattern": r"Incorrect integer value: '(.+)' for column", "対処方法": "数値変換処理を追加する", "エラータイプ": "型変換エラー" }, "Incorrect decimal value": { "pattern": r"Incorrect decimal value: '(.+)' for column", "対処方法": "数値変換処理を追加する", "エラータイプ": "型変換エラー" }, "cannot be null": { "pattern": r"Column '(.+)' cannot be null", "対処方法": "NULL値のチェックとデフォルト値の設定", "エラータイプ": "NULL制約違反" } } def get_table_definition(self, table_name: str) -> pd.DataFrame: """ テーブル定義を取得 Args: table_name: テーブル名 Returns: テーブル定義のDataFrame """ if table_name in self.table_definitions: return self.table_definitions[table_name] query = """ SELECT COLUMN_NAME, DATA_TYPE, COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT, CHARACTER_MAXIMUM_LENGTH, NUMERIC_PRECISION, NUMERIC_SCALE, COLUMN_COMMENT FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s ORDER BY ORDINAL_POSITION """ with self.connection.cursor() as cursor: cursor.execute(query, (table_name,)) columns = cursor.fetchall() df = pd.DataFrame(columns) self.table_definitions[table_name] = df return df def analyze_error(self, error_message: str) -> Dict[str, str]: """ エラーメッセージを分析 Args: error_message: エラーメッセージ Returns: 分析結果の辞書 """ result = { "エラータイプ": "不明", "対処方法": "エラー内容を確認してください", "エラー詳細": error_message } for error_type, rule in self.error_analysis_rules.items(): if error_type in error_message: match = re.search(rule["pattern"], error_message) if match: result["エラータイプ"] = rule["エラータイプ"] result["対処方法"] = rule["対処方法"] result["エラー詳細"] = match.group(1) if match.groups() else error_message break return result def get_column_info(self, table_name: str, column_name: str) -> Dict[str, Any]: """ カラム情報を取得 Args: table_name: テーブル名 column_name: カラム名 Returns: カラム情報の辞書 """ df = self.get_table_definition(table_name) column_row = df[df['COLUMN_NAME'] == column_name] if column_row.empty: return { "型": "不明", "コメント": "不明" } row = column_row.iloc[0] return { "型": row['COLUMN_TYPE'], "コメント": row['COLUMN_COMMENT'] if row['COLUMN_COMMENT'] else "" } def generate_error_report(self, error_logs: List[Dict], transplant_df: pd.DataFrame, mapping_df: pd.DataFrame = None) -> pd.DataFrame: """ エラーレポートを生成（統一フォーマット対応） Args: error_logs: エラー/警告ログのリスト（統一フォーマット） transplant_df: 移植データのDataFrame mapping_df: マッピング情報のDataFrame（オプション） Returns: エラーレポートのDataFrame """ report_data = [] # エラーがない場合でも最低限のレポートを作成 if not error_logs: report_data.append({ "エラー番号": "-", "YearNo": "-", "テーブル名": "エラーなし", "カラム名": "-", "エラータイプ": "正常終了", "エラー内容": "処理は正常に完了しました", "元の値": "-", "対処方法": "対処不要", "発生日時": datetime.now().strftime('%Y-%m-%d %H:%M:%S'), "重要度": "情報", "影響度": "なし" }) return pd.DataFrame(report_data) for idx, log_entry in enumerate(error_logs, 1): # 統一フォーマットからデータを取得 level = log_entry.get("level", "ERROR") table_name = log_entry.get("table", log_entry.get("テーブル名", "")) column_name = log_entry.get("column", log_entry.get("更新カラム名", "")) year_no = log_entry.get("year_no", log_entry.get("YearNo", "")) seitai_isyoku_id = log_entry.get("seitai_isyoku_id", "") message = log_entry.get("message", log_entry.get("エラー内容", "")) original_value = log_entry.get("original_value", log_entry.get("値", "")) expected_value = log_entry.get("expected_value", "") csv_source = log_entry.get("csv_source", "") error_detail = log_entry.get("error_detail", "") timestamp = log_entry.get("timestamp", datetime.now().strftime('%Y-%m-%d %H:%M:%S')) # カラム情報を取得 column_info = self.get_column_info(table_name, column_name) if table_name and column_name else {} # エラー分析（エラーメッセージがある場合） error_analysis = self.analyze_error(message) if message else { "エラータイプ": "警告" if level == "WARNING" else "エラー", "対処方法": "詳細を確認してください", "エラー詳細": error_detail or message } # CSVカラム名を推定（マッピング情報から） csv_column = csv_source if csv_source else self._find_csv_column(table_name, column_name, mapping_df) # 対処方法を詳細化 action = self._suggest_action(error_analysis, column_info, original_value) # 重要度の設定 severity = "警告" if level == "WARNING" else "エラー" report_row = { "NO": idx, "レベル": level, "CSV": csv_source or "項目名（移植時）", "CSVカラム": csv_column, "テーブル": self._get_table_japanese_name(table_name), "カラム名": column_name, "テーブル物理名": table_name, "物理カラム名": column_name, "型": column_info.get("型", ""), "コメント": column_info.get("コメント", ""), "移植登録ID(YearNo)": year_no, "生体移植ID": seitai_isyoku_id if seitai_isyoku_id else "-", "元の値": original_value, "期待値": expected_value if expected_value else "-", "エラー内容": message, "エラー詳細": error_analysis["エラー詳細"], "対処方法": action, "対応状況": "未対応", "エラータイプ": error_analysis["エラータイプ"], "重要度": severity, "発生日時": timestamp } report_data.append(report_row) return pd.DataFrame(report_data) def _find_csv_column(self, table_name: str, column_name: str, mapping_df: pd.DataFrame = None) -> str: """ マッピング情報からCSVカラム名を検索 Args: table_name: テーブル名 column_name: カラム名 mapping_df: マッピング情報 Returns: CSVカラム名 """ if mapping_df is None or mapping_df.empty: return "不明" # マッピング情報から検索（実装は実際のマッピング構造に依存） # ここでは簡易的な実装 return "不明" def _get_table_japanese_name(self, table_name: str) -> str: """テーブルの日本語名を取得""" table_name_map = { "T_ISHOKU_KIHON_LIV": "移植基本情報(共通-生体)", "T_DONOR_LIV": "ドナー情報(共通-生体)", "T_DONOR_KIDNEY_LIV": "ドナー情報(腎-生体)", "T_ISHOKU_KIHON_KIDNEY_LIV": "移植基本情報(腎-生体)", "T_LIVING_D_LIV": "生活状況D(生体-ドナー)", "T_KENSA_R_LIV": "検査項目R(生体-レシピエント)", "T_KENSA_D_LIV": "検査項目D(生体-ドナー)", "T_GAPPEI_R_LIV": "合併症R(生体-レシピエント)", "T_IJI_MENEKI_YOKUSEI_R_LIV": "維持期免疫抑制薬R(生体-レシピエント)", "T_KANSEN_R_LIV": "感染症R(生体-レシピエント)" } return table_name_map.get(table_name, table_name) def _suggest_action(self, error_analysis: Dict, column_info: Dict, error_data: Any) -> str: """ エラーに応じた具体的な対処方法を提案 Args: error_analysis: エラー分析結果 column_info: カラム情報 error_data: エラーデータ Returns: 対処方法の提案 """ error_type = error_analysis["エラータイプ"] column_type = column_info.get("型", "") if error_type == "データ長超過": # 現在の長さを解析 if "varchar" in column_type: match = re.search(r'varchar\((\d+)\)', column_type) if match: current_length = int(match.group(1)) required_length = len(str(error_data)) return f"varchar({current_length})をvarchar({required_length + 10})に拡張するか、データを{current_length}文字に切り詰める" elif "tinyint" in column_type and isinstance(error_data, (int, float)): if error_data > 255: return "tinyintをsmallintまたはintに変更する" return error_analysis["対処方法"] elif error_type == "型変換エラー": if "int" in column_type or "decimal" in column_type: return "数値以外の文字を除去する変換処理を追加" return error_analysis["対処方法"] elif error_type == "重複エラー": return "INSERT前に存在チェックを追加し、既存の場合はUPDATEまたはSKIP" return error_analysis["対処方法"] def save_error_report(self, report_df: pd.DataFrame, output_dir: str = "./logs", range_suffix: str = "") -> str: """ エラーレポートを保存（ERROR/WARNING別シート対応） Args: report_df: エラーレポートのDataFrame output_dir: 出力ディレクトリ range_suffix: ファイル名に追加する範囲サフィックス（分割実行対応） Returns: 保存したファイルパス """ os.makedirs(output_dir, exist_ok=True) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") # Excel形式で保存（分割実行対応） excel_path = os.path.join(output_dir, f"error_report_{timestamp}{range_suffix}.xlsx") with pd.ExcelWriter(excel_path, engine='openpyxl') as writer: # レベル列が存在する場合は、ERROR/WARNING別にシートを作成 if 'レベル' in report_df.columns: # ERROR専用シート error_df = report_df[report_df['レベル'] == 'ERROR'] if not error_df.empty: error_df.to_excel(writer, sheet_name='エラー一覧', index=False) self._adjust_column_width(writer, 'エラー一覧', error_df) # WARNING専用シート warning_df = report_df[report_df['レベル'] == 'WARNING'] if not warning_df.empty: warning_df.to_excel(writer, sheet_name='警告一覧', index=False) self._adjust_column_width(writer, '警告一覧', warning_df) # 全件一覧シート report_df.to_excel(writer, sheet_name='全件一覧', index=False) self._adjust_column_width(writer, '全件一覧', report_df) else: # 従来形式（レベル列なし） report_df.to_excel(writer, sheet_name='エラーレポート', index=False) self._adjust_column_width(writer, 'エラーレポート', report_df) # CSV形式でも保存（全件） csv_path = os.path.join(output_dir, f"error_report_{timestamp}{range_suffix}.csv") report_df.to_csv(csv_path, index=False, encoding='utf-8-sig') logger.info(f"エラーレポートを保存しました: {excel_path}") return excel_path def _adjust_column_width(self, writer, sheet_name: str, df: pd.DataFrame) -> None: """列幅を自動調整""" if sheet_name in writer.sheets: worksheet = writer.sheets[sheet_name] for column in df: try: column_length = max(df[column].astype(str).map(len).max(), len(column)) col_idx = df.columns.get_loc(column) worksheet.column_dimensions[chr(65 + col_idx)].width = min(column_length + 2, 50) except Exception: # カラムインデックスが範囲外の場合はスキップ pass def generate_summary_report(self, report_df: pd.DataFrame) -> pd.DataFrame: """ エラーサマリーレポートを生成（統一フォーマット対応） Args: report_df: エラーレポートのDataFrame Returns: サマリーレポートのDataFrame """ summary_data = [] # レベル別集計（ERROR/WARNING） if 'レベル' in report_df.columns: level_summary = report_df.groupby('レベル').size().reset_index(name='件数') level_summary['分類'] = 'レベル別' summary_data.append(level_summary.rename(columns={'レベル': '項目'})) # テーブル別集計 if 'テーブル物理名' in report_df.columns and 'エラータイプ' in report_df.columns: table_summary = report_df.groupby(['テーブル物理名', 'エラータイプ']).size().reset_index(name='件数') table_summary['分類'] = 'テーブル別' table_summary['項目'] = table_summary['テーブル物理名'] + ' - ' + table_summary['エラータイプ'] summary_data.append(table_summary[['分類', '項目', '件数']]) # エラータイプ別集計 if 'エラータイプ' in report_df.columns: error_type_summary = report_df.groupby('エラータイプ').agg({ 'NO': 'count', '対処方法': lambda x: x.value_counts().index[0] if len(x) > 0 else '' }).reset_index() error_type_summary.columns = ['エラータイプ', '件数', '主な対処方法'] error_type_summary['分類'] = 'エラータイプ別' error_type_summary['項目'] = error_type_summary['エラータイプ'] summary_data.append(error_type_summary[['分類', '項目', '件数', '主な対処方法']]) # 統合サマリー作成 if summary_data: final_summary = pd.concat(summary_data, ignore_index=True) # 列を統一 required_columns = ['分類', '項目', '件数', '主な対処方法'] for col in required_columns: if col not in final_summary.columns: final_summary[col] = '-' return final_summary[required_columns] else: # フォールバック（従来形式） if 'エラータイプ' in report_df.columns: error_type_summary = report_df.groupby('エラータイプ').agg({ 'NO': 'count', '対処方法': lambda x: x.value_counts().index[0] if len(x) > 0 else '' }).reset_index() error_type_summary.columns = ['エラータイプ', '件数', '主な対処方法'] return error_type_summary else: return pd.DataFrame({'分類': ['情報'], '項目': ['データなし'], '件数': [0], '主な対処方法': ['-']})