Survey Insight MCP Server

"""CSVファイルの読み込みと前処理モジュール.""" import logging from pathlib import Path from typing import Optional import pandas as pd logger = logging.getLogger(__name__) class CSVLoader: """CSVファイルの読み込みと前処理を行うクラス. アンケートCSVファイルを読み込み、エンコーディング自動検出、 データクレンジング、自由コメント列の自動識別を行います。 """ def __init__(self): """初期化.""" # ExcelのCSV（ANSI/Shift-JIS）を優先的に試す self.encodings = ["cp932", "shift-jis", "utf-8-sig", "utf-8"] def load(self, csv_path: str, comment_column: Optional[str] = None) -> tuple[pd.DataFrame, str]: """CSVファイルを読み込む. Args: csv_path: CSVファイルのパス comment_column: 自由コメント列の名前（省略時は自動検出） Returns: tuple[pd.DataFrame, str]: (読み込んだDataFrame, コメント列名) Raises: FileNotFoundError: ファイルが存在しない場合 ValueError: CSVファイルの読み込みに失敗した場合 """ csv_file = Path(csv_path) if not csv_file.exists(): raise FileNotFoundError(f"CSVファイルが見つかりません: {csv_path}") # エンコーディング自動検出して読み込み encoding = self.detect_encoding(csv_path) logger.info(f"CSVファイル読み込み: {csv_path} (エンコーディング: {encoding})") try: df = pd.read_csv(csv_path, encoding=encoding) except Exception as e: raise ValueError(f"CSVファイルの読み込みに失敗しました: {e}") # データクレンジング df = self.clean_data(df) # コメント列の検出 if comment_column is None: comment_column = self.detect_comment_column(df) logger.info(f"コメント列を自動検出: {comment_column}") elif comment_column not in df.columns: raise ValueError(f"指定されたコメント列が見つかりません: {comment_column}") logger.info(f"CSVファイル読み込み完了: {len(df)}行") return df, comment_column def detect_encoding(self, csv_path: str) -> str: """エンコーディング自動検出. Args: csv_path: CSVファイルのパス Returns: str: 検出されたエンコーディング Raises: ValueError: すべてのエンコーディングで読み込みに失敗した場合 """ for encoding in self.encodings: try: # ファイルを読み込んでCSVとしてパースできるかテスト df_test = pd.read_csv(csv_path, encoding=encoding, nrows=5) # パースに成功し、データが取得できた場合 if not df_test.empty and len(df_test.columns) > 0: logger.info(f"エンコーディング検出成功: {encoding}") return encoding except (UnicodeDecodeError, pd.errors.ParserError, Exception) as e: logger.debug(f"エンコーディング {encoding} でのパース失敗: {e}") continue raise ValueError(f"CSVファイルのエンコーディングを検出できませんでした: {csv_path}") def detect_comment_column(self, df: pd.DataFrame) -> str: """自由コメント列を自動検出. Args: df: DataFrame Returns: str: コメント列名 Raises: ValueError: コメント列が見つからない場合 """ # コメント列の候補となるキーワード comment_keywords = ["コメント", "comment", "自由記述", "意見", "感想", "フィードバック"] # カラム名に候補キーワードが含まれているか確認 for col in df.columns: col_lower = str(col).lower() for keyword in comment_keywords: if keyword.lower() in col_lower: return col # 見つからない場合は、最も長い文字列を含むカラムを選択 max_length_col = None max_avg_length = 0 for col in df.columns: if df[col].dtype == object: # 文字列型のカラムのみ avg_length = df[col].astype(str).str.len().mean() if avg_length > max_avg_length: max_avg_length = avg_length max_length_col = col if max_length_col is None: raise ValueError("コメント列を検出できませんでした。comment_columnパラメータで明示的に指定してください。") return max_length_col def clean_data(self, df: pd.DataFrame) -> pd.DataFrame: """データクレンジング. Args: df: DataFrame Returns: pd.DataFrame: クレンジング済みDataFrame """ # 完全に空の行を削除 df = df.dropna(how="all") # 重複行を削除 df = df.drop_duplicates() # インデックスをリセット df = df.reset_index(drop=True) # 文字列型カラムの前後空白を削除 for col in df.columns: if df[col].dtype == object: df[col] = df[col].astype(str).str.strip() # 'nan'文字列をNaNに変換 df[col] = df[col].replace("nan", pd.NA) logger.info(f"データクレンジング完了: {len(df)}行") return df