Survey Insight MCP Server

"""Survey Insight MCPサーバ実装.""" import logging import os from pathlib import Path from typing import Any, Optional import pandas as pd from dotenv import load_dotenv from mcp.server import Server from mcp.types import TextContent, Tool from survey_insight.ai_analyzer import AIAnalyzer from survey_insight.axis_analyzer import AxisAnalyzer from survey_insight.chart_generator import ChartGenerator from survey_insight.csv_loader import CSVLoader from survey_insight.html_generator import HTMLGenerator from survey_insight.text_analyzer import TextAnalyzer from survey_insight.wordcloud_generator import WordCloudGenerator # .envファイルから環境変数を読み込み env_path = Path.cwd() / ".env" if env_path.exists(): load_dotenv(env_path) logger = logging.getLogger(__name__) def get_output_dir() -> Path: """outputディレクトリのパスを取得（カレントディレクトリ基準）.""" # MCPサーバ実行時、カレントディレクトリはプロジェクトルート output_dir = Path.cwd() / "output" output_dir.mkdir(parents=True, exist_ok=True) logger.info(f"Output directory: {output_dir.absolute()}") return output_dir.absolute() # グローバルなサーバーインスタンス app = Server("survey-insight") # 分析コンポーネント logger.info("Initializing survey-insight components...") try: csv_loader = CSVLoader() logger.info("✓ CSVLoader initialized") text_analyzer = TextAnalyzer() logger.info("✓ TextAnalyzer initialized") axis_analyzer = AxisAnalyzer() logger.info("✓ AxisAnalyzer initialized") wordcloud_gen = WordCloudGenerator() logger.info("✓ WordCloudGenerator initialized") chart_gen = ChartGenerator() logger.info("✓ ChartGenerator initialized") ai_analyzer = AIAnalyzer() logger.info("✓ AIAnalyzer initialized") html_gen = HTMLGenerator() logger.info("✓ HTMLGenerator initialized") logger.info("All components initialized successfully!") except Exception as e: logger.error(f"Component initialization failed: {e}", exc_info=True) raise @app.list_tools() async def list_tools() -> list[Tool]: """利用可能なツール一覧を返す.""" return [ Tool( name="analyze_survey", description="アンケートCSVファイルを分析し、形態素解析、WordCloud、グラフを含む洗練されたHTMLレポートを生成します", inputSchema={ "type": "object", "properties": { "csv_path": {"type": "string", "description": "アンケートCSVファイルのパス"}, "comment_column": {"type": "string", "description": "自由コメント列の名前（省略時は自動検出）"}, "analysis_axes": { "type": "array", "items": {"type": "string"}, "description": "分析軸のカラム名リスト（例: ['部署', '年代', '役職']）", }, "output_path": { "type": "string", "description": "HTMLレポートの出力パス（デフォルト: output/survey_report.html）", }, "enable_ai_analysis": {"type": "boolean", "description": "AI課題分析を有効化（デフォルト: true）"}, "language": { "type": "string", "enum": ["ja", "en"], "description": "言語コード（省略時はコメント列から自動判別）", }, }, "required": ["csv_path"], }, ), Tool( name="update_ai_analysis", description="""既存のHTMLレポートにClaude Codeが分析した課題と解決策を追加します。 使用前に必ず以下を実施してください： 1. analysis_summary.txtを読み込み（グラフデータ+分析軸別コメント詳細を含む） 2. グラフデータ（頻出キーワード、分析軸別統計）を定量的に分析 3. 分析軸別コメント詳細でセグメント特性を把握（全コメントが含まれる） 4. 定量データと定性データを統合して課題を抽出 5. 具体的なコメント引用を含めた詳細な課題説明を作成 APIキーなしでClaude Codeサブスクリプションのみで利用する場合に使用します。""", inputSchema={ "type": "object", "properties": { "report_path": {"type": "string", "description": "更新対象のHTMLレポートのパス"}, "issues": { "type": "array", "items": { "type": "object", "properties": { "title": {"type": "string", "description": "課題のタイトル"}, "detail": {"type": "string", "description": "課題の詳細説明"}, "priority": {"type": "string", "enum": ["高", "中", "低"], "description": "優先度"}, "impact": {"type": "string", "description": "影響範囲（オプション）"}, }, "required": ["title", "detail", "priority"], }, "description": "検出された課題のリスト", }, "solutions": { "type": "array", "items": { "type": "object", "properties": { "actions": { "type": "array", "items": {"type": "string"}, "description": "アクションプランのリスト", }, "difficulty": {"type": "string", "description": "実装難易度（オプション）"}, "period": {"type": "string", "description": "実施期間（オプション）"}, "expected_effect": {"type": "string", "description": "期待効果（オプション）"}, }, }, "description": "改善提案のリスト", }, }, "required": ["report_path", "issues"], }, ), Tool( name="generate_wordcloud", description="テキストデータからWordCloudを生成", inputSchema={ "type": "object", "properties": { "text": {"type": "string", "description": "WordCloud生成元のテキスト"}, "output_path": {"type": "string", "description": "画像出力パス（デフォルト: output/wordcloud.png）"}, "color_scheme": { "type": "string", "enum": ["accenture", "default"], "description": "カラースキーム", }, }, "required": ["text"], }, ), Tool( name="extract_keywords", description="形態素解析でキーワードを抽出", inputSchema={ "type": "object", "properties": { "text": {"type": "string", "description": "解析対象のテキスト"}, "top_n": {"type": "integer", "description": "上位N件のキーワードを返す（デフォルト: 20）"}, }, "required": ["text"], }, ), ] @app.call_tool() async def call_tool(name: str, arguments: dict) -> list[TextContent]: """ツールを実行.""" logger.info(f"Tool called: {name} with arguments: {arguments.keys()}") try: if name == "analyze_survey": logger.info("Starting analyze_survey...") # outputディレクトリを取得 output_dir = get_output_dir() result = await handle_analyze_survey( csv_path=arguments["csv_path"], comment_column=arguments.get("comment_column"), analysis_axes=arguments.get("analysis_axes"), output_path=arguments.get("output_path", str(output_dir / "survey_report.html")), enable_ai_analysis=arguments.get("enable_ai_analysis", True), language=arguments.get("language"), ) logger.info("analyze_survey completed successfully") return [TextContent(type="text", text=result)] elif name == "generate_wordcloud": logger.info("Starting generate_wordcloud...") # outputディレクトリを取得 output_dir = get_output_dir() result = await handle_generate_wordcloud( text=arguments["text"], output_path=arguments.get("output_path", str(output_dir / "wordcloud.png")), color_scheme=arguments.get("color_scheme", "accenture"), ) logger.info("generate_wordcloud completed successfully") return [TextContent(type="text", text=result)] elif name == "extract_keywords": logger.info("Starting extract_keywords...") result = await handle_extract_keywords( text=arguments["text"], top_n=arguments.get("top_n", 20), ) logger.info("extract_keywords completed successfully") return [TextContent(type="text", text=result)] elif name == "update_ai_analysis": logger.info("Starting update_ai_analysis...") result = await handle_update_ai_analysis( report_path=arguments["report_path"], issues=arguments["issues"], solutions=arguments.get("solutions", []), ) logger.info("update_ai_analysis completed successfully") return [TextContent(type="text", text=result)] else: error_msg = f"Unknown tool: {name}" logger.error(error_msg) return [TextContent(type="text", text=error_msg)] except Exception as e: import traceback error_details = traceback.format_exc() logger.error(f"Tool execution error in {name}: {e}\n{error_details}") return [TextContent(type="text", text=f"Error executing {name}: {str(e)}\n\nDetails:\n{error_details}")] def _detect_analysis_axes(df, comment_column: str) -> list[str]: """分析軸候補を自動検出. Args: df: DataFrame comment_column: コメント列名 Returns: list[str]: 検出された分析軸のリスト """ axes = [] exclude_columns = [comment_column, "ID", "id", "Id"] for col in df.columns: # 除外対象のカラムはスキップ if col in exclude_columns: continue # ユニーク値数を確認 unique_count = df[col].nunique() total_count = len(df) # カテゴリカルデータの条件： # - ユニーク値が2以上 # - ユニーク値が全体の50%未満（カテゴリとして意味がある） # - データ型がobjectまたはcategory if ( unique_count >= 2 and unique_count < total_count * 0.5 and (df[col].dtype == "object" or pd.api.types.is_categorical_dtype(df[col])) ): axes.append(col) logger.info(f"分析軸候補を検出: {col} (ユニーク値数: {unique_count})") return axes def _generate_analysis_summary( csv_path: str, df: Any, keywords: list[dict], axes_analysis: dict, analysis_axes: list[str], comment_column: str, ) -> str: """AI診断用の分析サマリーテキストを生成. Args: csv_path: CSVファイルパス df: DataFrame keywords: キーワードリスト axes_analysis: 軸別分析結果 analysis_axes: 分析軸リスト comment_column: コメント列名 Returns: str: AI診断用のサマリーテキスト """ summary = f"""# {Path(csv_path).stem} - Survey Analysis Summary ## 📊 基本統計情報 - 総回答数: {len(df)}件 - 分析対象列: {comment_column} - 抽出キーワード数: {len(keywords)}個 - 分析軸数: {len(analysis_axes) if analysis_axes else 0}個 ## 🔑 頻出キーワード Top 20 """ for i, kw in enumerate(keywords[:20], 1): summary += f"{i:2d}. {kw['keyword']:15s} - {kw['count']:3d}回 (スコア: {kw['score']:.2f})\n" # 分析軸別の詳細情報 if axes_analysis and analysis_axes: summary += "\n## 📈 分析軸別データ\n\n" for axis_name in analysis_axes: axis_result = axes_analysis["axes_results"].get(axis_name) if not axis_result: continue summary += f"### {axis_name}別分析\n\n" categories = axis_result["categories"] # カテゴリ別統計（回答数と割合） total_count = sum(cat_data["count"] for cat_data in categories.values()) summary += "**カテゴリ別回答数:**\n\n" for cat_name, cat_data in categories.items(): count = cat_data["count"] percentage = (count / total_count * 100) if total_count > 0 else 0 summary += f"- {cat_name}: {count}件 ({percentage:.1f}%)\n" # カテゴリ別Top 5キーワード summary += f"\n**{axis_name}別 Top 5キーワード:**\n\n" for cat_name, cat_data in categories.items(): summary += f"**[{cat_name}]**\n" # top_keywordsはタプルのリスト: [(keyword, count), ...] top_keywords = cat_data.get("top_keywords", [])[:5] if top_keywords: for j, (keyword, count) in enumerate(top_keywords, 1): summary += f" {j}. {keyword} ({count}回)\n" else: summary += " （該当なし）\n" summary += "\n" # 分析軸別コメント（全コメントがカテゴリ別に分類される） if axes_analysis and analysis_axes: summary += "\n## 📋 分析軸別コメント詳細\n\n" for axis_name in analysis_axes: axis_result = axes_analysis["axes_results"].get(axis_name) if not axis_result: continue summary += f"### {axis_name}別コメント\n\n" categories = axis_result["categories"] for cat_name, cat_data in categories.items(): # カテゴリに属するコメントを抽出 cat_df = df[df[axis_name] == cat_name] cat_comments = cat_df[comment_column].dropna().astype(str).tolist() count = len(cat_comments) summary += f"**[{cat_name}]** ({count}件)\n\n" for i, comment in enumerate(cat_comments, 1): summary += f"{i}. {comment}\n" summary += "\n" summary += f"\n---\n生成日時: {pd.Timestamp.now().strftime('%Y-%m-%d %H:%M:%S')}\n" return summary async def handle_analyze_survey( csv_path: str, comment_column: Optional[str] = None, analysis_axes: Optional[list[str]] = None, output_path: str = "output/survey_report.html", enable_ai_analysis: bool = True, language: Optional[str] = None, ) -> str: """アンケート分析メインハンドラ. Args: csv_path: CSVファイルパス comment_column: コメント列名 analysis_axes: 分析軸リスト output_path: 出力HTMLパス enable_ai_analysis: AI分析有効化 language: 言語コード ("ja" or "en"、省略時は自動判別） Returns: str: 分析結果のサマリー """ logger.info("=" * 80) logger.info("🚀 [VERSION CHECK] mcp_server.py - CODE VERSION: 2024-10-25-v2-BASE64") logger.info("=" * 80) logger.info(f"アンケート分析開始: {csv_path}") # AI分析モードの判定（優先順位: USE_CLAUDE_CODE_SUBSCRIPTION > LLM_API_KEY） use_claude_code_subscription = os.getenv("USE_CLAUDE_CODE_SUBSCRIPTION", "false").lower() == "true" # LLMプロバイダーとAPIキーの取得（後方互換性のためANTHROPIC_API_KEYもチェック） llm_provider = os.getenv("LLM_PROVIDER", "anthropic") llm_api_key = os.getenv("LLM_API_KEY") or os.getenv("ANTHROPIC_API_KEY") # APIキーが有効かチェック（ダミー値と重複プレフィックスを除外） is_valid_api_key = ( llm_api_key and llm_api_key not in ["your_api_key_here", "", "None", "none"] and not llm_api_key.startswith("ANTHROPIC_API_KEY=") and not llm_api_key.startswith("LLM_API_KEY=") ) # ロジック: サブスクリプション優先 > APIキー > 無効化 if use_claude_code_subscription: # USE_CLAUDE_CODE_SUBSCRIPTION=trueが最優先（Claude Codeサブスクリプションを使用） logger.info("✓ Claude Code Subscription モードが有効です") enable_ai_analysis = False # API経由の分析を無効化 elif is_valid_api_key: # 有効なAPIキーがある場合はAPI経由でAI分析を実行 logger.info(f"✓ {llm_provider.upper()} API経由でAI分析を実行します") logger.info(f" APIキープレフィックス: {llm_api_key[:20]}...") enable_ai_analysis = enable_ai_analysis # 引数の値を尊重 else: # どちらも設定されていない、またはAPIキーが無効な場合 if llm_api_key: logger.error(f"✗ APIキーが無効です: {llm_api_key[:30]}...") logger.error(" .envファイルのLLM_API_KEYの設定を確認してください") logger.error(" 正しい形式: LLM_PROVIDER=anthropic, LLM_API_KEY=sk-ant-api03-...") else: logger.warning("✗ 有効なAPIキーもClaude Code Subscriptionも設定されていません") logger.warning(" AI分析は無効化されます") enable_ai_analysis = False # 0. outputディレクトリを取得 output_dir = get_output_dir() logger.info(f"Using output directory: {output_dir}") # 1. CSV読み込み df, detected_comment_column = csv_loader.load(csv_path, comment_column) comment_column = detected_comment_column # 1.5. 分析軸の検証と自動検出 if analysis_axes: # 指定された分析軸がCSVに存在するか検証 invalid_axes = [axis for axis in analysis_axes if axis not in df.columns] if invalid_axes: logger.warning(f"指定された分析軸がCSVに存在しません: {invalid_axes}") logger.warning(f"利用可能なカラム: {df.columns.tolist()}") logger.warning("分析軸を自動検出に切り替えます。") analysis_axes = None if not analysis_axes: analysis_axes = _detect_analysis_axes(df, comment_column) if not analysis_axes: logger.warning("分析軸が検出できませんでした。軸別分析をスキップします。") else: logger.info(f"自動検出した分析軸: {analysis_axes}") # 2. 言語検出（手動指定 or 自動判別） all_comments = " ".join(df[comment_column].dropna().astype(str).tolist()) if language: detected_language = language logger.info(f"言語が手動で指定されました: {detected_language}") else: # コメント列全体から言語を自動判別 detected_language = text_analyzer.detect_language(all_comments) logger.info(f"言語を自動判別しました: {detected_language}") # 3. 形態素解析（言語を明示的に指定） keywords = text_analyzer.extract_keywords(all_comments, top_n=50, language=detected_language) # 4. WordCloud用の複合名詞+単語頻度辞書を生成（言語を明示的に指定） wordcloud_frequencies = text_analyzer.extract_compound_keywords( all_comments, min_frequency=2, max_compound_length=3, compound_ratio=0.7, language=detected_language ) # 4. WordCloud生成（外部ファイルとして保存） wordcloud_path = str(output_dir / "wordcloud.png") if wordcloud_frequencies: wordcloud_gen.generate(wordcloud_frequencies, wordcloud_path, color_scheme="accenture") else: logger.warning("WordCloud用のキーワードが抽出できませんでした。元のテキストで生成します。") wordcloud_gen.generate(all_comments, wordcloud_path, color_scheme="accenture") logger.info(f"✅ WordCloud画像を生成しました: {wordcloud_path}") # 5. グラフ生成 keyword_dict = {kw["keyword"]: kw["count"] for kw in keywords} bar_chart = chart_gen.create_bar_chart(keyword_dict, "頻出キーワードランキング", top_n=20) # 6. 分析軸処理 axes_charts = {} axes_analysis = None if analysis_axes: axes_analysis = axis_analyzer.compare_axes(df, keywords, analysis_axes, comment_column) for axis_name, result in axes_analysis["axes_results"].items(): # 軸別の円グラフ生成 categories = result["categories"] category_counts = {cat: data["count"] for cat, data in categories.items()} axes_charts[axis_name] = chart_gen.create_pie_chart(category_counts, f"{axis_name}別分布") # 7. AI分析 issues = [] solutions = [] ai_warning = None if use_claude_code_subscription: # Claude Code Subscriptionモードの場合は警告なし（後でClaude Codeが分析） ai_warning = None elif not is_valid_api_key: # 有効なAPIキーがない場合の警告 ai_warning = ( "LLM_API_KEYが設定されていないか、無効な値です。AI分析を実行するには、" ".envファイルにLLM_PROVIDER=anthropic, LLM_API_KEY=your_api_keyを設定するか、" "USE_CLAUDE_CODE_SUBSCRIPTION=trueを設定してください。" ) elif enable_ai_analysis: # API分析実行 comments_sample = df[comment_column].dropna().astype(str).tolist()[:50] ai_result = await ai_analyzer.analyze_issues(keywords, axes_analysis, comments_sample) issues = ai_result.get("issues", []) ai_warning = ai_result.get("warning") if issues: solutions = await ai_analyzer.generate_solutions(issues) # 8. HTML生成 # AI情報の生成 ai_info = None if use_claude_code_subscription: ai_info = "Claude Code Subscription" elif is_valid_api_key and ai_analyzer.model: # プロバイダー名を大文字に変換して表示 provider_name = llm_provider.capitalize() if llm_provider == "anthropic" else llm_provider.upper() ai_info = f"{provider_name} {ai_analyzer.model}" # WordCloud画像は外部ファイル参照（HTMLファイルサイズを抑えてClaude Code読み込みを確実にするため） context = { "report_title": f"{Path(csv_path).stem} 分析レポート", "total_responses": len(df), "total_keywords": len(keywords), "total_issues": len(issues), "total_axes": len(analysis_axes) if analysis_axes else 0, "wordcloud_image": Path(wordcloud_path).name, # 外部ファイル参照 "bar_chart": bar_chart, "axes_charts": axes_charts, "issues": issues, "solutions": solutions, "ai_warning": ai_warning, "ai_info": ai_info, } html_path = html_gen.generate("dashboard.html", context, output_path) # 9. AI診断用サマリーファイル生成（全モードで生成） summary_path = None # AI診断用の軽量サマリーファイルを常に生成 summary_text = _generate_analysis_summary(csv_path, df, keywords, axes_analysis, analysis_axes, comment_column) summary_path = str(output_dir / "analysis_summary.txt") with open(summary_path, "w", encoding="utf-8") as f: f.write(summary_text) logger.info(f"✅ AI診断用サマリーファイルを生成しました: {summary_path}") if use_claude_code_subscription: logger.info(f" ファイルサイズ: {len(summary_text)} bytes (HTMLの代わりにこちらを読み込み)") else: logger.info(f" ファイルサイズ: {len(summary_text)} bytes (参考資料として利用可能)") # Claude Code Subscription モードの場合、AI分析専用メッセージを返す if use_claude_code_subscription: summary = f""" 🤖 Survey Analysis Complete - AI Analysis Required =================================================== ✅ Files Generated: - HTML Report: {html_path} - WordCloud Image: {wordcloud_path} - Analysis Summary: {summary_path} ⭐ (AI診断用) 📊 Basic Statistics: - {len(df)} responses analyzed - {len(keywords)} keywords extracted - {len(analysis_axes) if analysis_axes else 0} analysis axes ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ ⚠️ IMPORTANT: Read the Summary File, NOT HTML ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ HTMLファイル ({Path(html_path).stat().st_size / 1024:.1f}KB) は大きすぎて 読み込みエラーが発生します。代わりに軽量なサマリーファイルを使用してください。 Your Task - 総合的なアンケート分析: 1. Read ONLY the analysis summary file: 📄 {summary_path} このファイルには以下が含まれています： - 基本統計情報（回答数、キーワード数、分析軸数） - 頻出キーワード Top 20（全体傾向の把握） - 分析軸別データ（カテゴリ別回答数・割合・上位キーワード） - **分析軸別コメント詳細**（各カテゴリの全コメント、重複なし） 2. 総合的な分析アプローチ: **Step 1: グラフデータの定量分析** - 頻出キーワードランキングから全体傾向を把握 - 分析軸別のカテゴリ分布（回答数・割合）を確認 - 各カテゴリの上位キーワードから特徴を抽出 **Step 2: 分析軸別コメントの質的分析** - 分析軸別コメント詳細で各セグメントの特性を理解（全コメントが含まれる） - グラフの数値と実際のコメント内容を照合 - セグメント間の違いや共通点を発見 **Step 3: 統合分析** - 定量データ（グラフ）と定性データ（コメント）を統合 - 潜在的な課題やニーズを抽出 3. 3～5つの重要課題を特定: - グラフデータと全コメントの両方を根拠にする - 具体的なコメント引用で裏付ける - セグメント別の傾向差も考慮する - **必ず日本語で記述** {{ "title": "課題のタイトル（日本語）", "detail": "具体的なコメント例を含む詳細説明（日本語）", "priority": "高/中/低", "impact": "影響範囲（任意・日本語）" }} 4. 2～4つの解決策を生成: - 発見された課題に対する具体的な改善策 - セグメント別のアプローチも検討 - **必ず日本語で記述** {{ "actions": ["アクション1（日本語）", "アクション2（日本語）", ...], "difficulty": "実装難易度（任意・日本語）", "period": "実施期間（任意・日本語）", "expected_effect": "期待される効果（任意・日本語）" }} 5. update_ai_analysis ツールで結果を追加: - report_path: "{html_path}" - issues: [課題の辞書リスト（日本語）] - solutions: [解決策の辞書リスト（日本語）] ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ ⛔ FORBIDDEN ACTIONS: ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ - Using Bash tool with python/pandas commands - Reading CSV file directly - Using any programming language to analyze data ✅ ONLY ALLOWED: - Read tool for HTML file - update_ai_analysis tool for adding results """ else: # 通常モード（APIキー使用 or AI分析無効）のサマリー summary = f""" Survey Insight Analysis Complete! ================================== 📊 Analysis Summary: - Total Responses: {len(df)} - Keywords Extracted: {len(keywords)} - Issues Detected: {len(issues)} - Analysis Axes: {len(analysis_axes) if analysis_axes else 0} 📁 Output Files: - HTML Report: {html_path} - WordCloud: {wordcloud_path} 🔝 Top 10 Keywords: """ for i, kw in enumerate(keywords[:10], 1): summary += f"{i}. {kw['keyword']}: {kw['count']}回\n" if issues: summary += "\n⚠️ Priority Issues:\n" for i, issue in enumerate([iss for iss in issues if iss.get("priority") == "高"][:5], 1): summary += f"{i}. {issue.get('title', 'N/A')}\n" elif not is_valid_api_key: summary += "\n⚠️ AI Analysis Disabled:\n" summary += "有効なAPIキーが設定されていません。AI分析を実行するには:\n" summary += "- Option 1: .envファイルにLLM_PROVIDER=anthropic, LLM_API_KEY=your_api_keyを設定\n" summary += "- Option 2: USE_CLAUDE_CODE_SUBSCRIPTION=trueを設定\n" logger.info("アンケート分析完了") return summary async def handle_generate_wordcloud( text: str, output_path: str = "output/wordcloud.png", color_scheme: str = "accenture" ) -> str: """WordCloud生成ハンドラ. Args: text: テキスト output_path: 出力パス color_scheme: カラースキーム Returns: str: 結果メッセージ """ logger.info("WordCloud生成開始") path = wordcloud_gen.generate(text, output_path, color_scheme) return f"WordCloud generated: {path}" async def handle_extract_keywords(text: str, top_n: int = 20) -> str: """キーワード抽出ハンドラ. Args: text: テキスト top_n: 上位N件 Returns: str: JSON形式のキーワードリスト """ logger.info(f"キーワード抽出開始: top_n={top_n}") keywords = text_analyzer.extract_keywords(text, top_n) result = "Extracted Keywords:\n" result += "==================\n\n" for i, kw in enumerate(keywords, 1): result += f"{i}. {kw['keyword']}: {kw['count']}回 (score: {kw['score']:.2f})\n" return result async def handle_update_ai_analysis(report_path: str, issues: list[dict], solutions: list[dict]) -> str: """AI分析結果をHTMLレポートに追加. Args: report_path: HTMLレポートのパス issues: 課題リスト solutions: 解決策リスト Returns: str: 結果メッセージ """ logger.info(f"AI分析結果の追加開始: {report_path}") # HTMLレポートを更新 updated_path = html_gen.update_ai_analysis(report_path, issues, solutions) result = f""" AI Analysis Updated Successfully! ================================== 📊 Updated Report: {updated_path} ✅ Issues Added: {len(issues)} ✅ Solutions Added: {len(solutions)} Priority Breakdown: """ # 優先度ごとの集計 priority_count = {"高": 0, "中": 0, "低": 0} for issue in issues: priority = issue.get("priority", "中") priority_count[priority] = priority_count.get(priority, 0) + 1 result += f"- 高: {priority_count['高']}件\n" result += f"- 中: {priority_count['中']}件\n" result += f"- 低: {priority_count['低']}件\n" logger.info("AI分析結果の追加完了") return result