improve some tools.

app.py

@@ -2,8 +2,13 @@ import gradio as gr
 import pandas as pd
 import sys
 import os
+import json
 
-from utils.tools import get_kb, get_schema, get_tables, get_meaning, execute_sqlite_query
+from utils.tools import get_kb, get_schema, get_tables, get_meaning, execute_sqlite_query, search_meaning
+
+
+def get_message(msg):
+    return {"status": "error", "headers":['message'], "data": [[f"{msg}"]]}
 
 @gr.mcp.tool()
 def get_all_databases() -> list:
@@ -29,34 +34,63 @@ def kb_query(db_name, knowledge_keyword):
     Query the knowledge base for a specific database with optional keyword filtering.
     
     This function retrieves knowledge base information for a specified database.
-    If a keyword is provided, it filters the results based on that keyword.
+    If keywords are provided, it filters the results based on those keywords.
+    Multiple keywords can be separated by commas, 'or', or spaces.
     
     Args:
         db_name (str): The name of the database to query. Must not be empty.
-        knowledge_keyword (str): Optional keyword to filter knowledge base results.
+        knowledge_keyword (str): Optional keywords to filter knowledge base results.
+                               Multiple keywords can be separated by commas, 'or', or spaces.
                                If empty or None, returns all knowledge for the database.
     
     Returns:
-        pandas.DataFrame: Query results containing knowledge base data, or error message
-                         if no database is selected or no results found.
+        dict: Dictionary containing query results with structure:
+              - Success: {"status": "success", "headers": [...], "data": [...]}
+              - Error: {"status": "error", "headers": ['message'], "data": [['Error message.']]}
     
     Example:
-        >>> result = kb_query("sales_db", "customer")
+        >>> result = kb_query("sales_db", "customer, order")
         >>> print(result)
-        # Returns DataFrame with customer-related knowledge from sales_db
+        # Returns dict with customer and order-related knowledge from sales_db
     """
     if not db_name:
-        return pd.DataFrame({"message": ["請先選擇資料庫"]})
+        return get_message("Please select db_name.")
     
     if not knowledge_keyword:
         result = get_kb(db_name)
     else:
-        result = get_kb(db_name, knowledge_keyword)
-    
+        # Split keywords by comma, 'or', or space
+        keywords = [k.strip() for k in knowledge_keyword.replace('or', ',').replace(' ', ',').split(',') if k.strip()]
+
+        # Get results for each keyword and combine them
+        combined_results = []
+        for keyword in keywords:
+            keyword_result = get_kb(db_name, keyword)
+            # 使用 append 而不是 extend,因為 get_kb 返回 DataFrame
+            if not keyword_result.empty:
+                combined_results.append(keyword_result)
+    
+        # Remove duplicates if any
+        # Convert DataFrames to a single DataFrame and remove duplicate rows
+        if combined_results:
+            result = pd.concat(combined_results, ignore_index=True).drop_duplicates()
+        else:
+            result = pd.DataFrame()
+
     if len(result) == 0:
-        return pd.DataFrame({"message": ["沒有找到相關知識"]})
-    
-    return result
+        return get_message("Not Found.")
+    
+    # Convert DataFrame to JSON format with columns and rows
+    data = {
+        "columns": result.columns.tolist(),
+        "rows": result.values.tolist()
+    }
+    return {"status": "success", "headers": result.columns.tolist(), "data":result.values.tolist()}
+    # return json.dumps(
+    #     {"status": "success", "data": data},
+    #     ensure_ascii=False,
+    #     indent=2
+    # )
 
 def schema_query(db_name, table_name):
     """
@@ -70,23 +104,29 @@ def schema_query(db_name, table_name):
         table_name (str): The name of the table to query schema for. Must not be empty.
     
     Returns:
-        pandas.DataFrame: Query results containing table schema information, or error message
-                         if parameters are missing or no schema found.
+        dict: Dictionary containing query results with structure:
+              - Success: {"status": "success", "headers": [...], "data": [...]}
+              - Error: {"status": "error", "headers": ['message'], "data": [['Error message.']]}
     
     Example:
         >>> result = schema_query("sales_db", "customers")
         >>> print(result)
-        # Returns DataFrame with column definitions for customers table
+        # Returns dict with column definitions for customers table
     """
     if not db_name or not table_name:
-        return pd.DataFrame({"message": ["請選擇資料庫和資料表"]})
+        return get_message("請選擇資料庫和資料表")
     
-    result = get_schema(db_name, table_name)
+    # 取得 schema 資訊
+    schema_result = get_schema(db_name, table_name)
     
-    if len(result) == 0:
-        return pd.DataFrame({"message": ["沒有找到相關資料表結構"]})
+    if len(schema_result) == 0:
+        return get_message("Not Found.")
     
-    return result
+    return {
+        "status": "success",
+        "headers": schema_result.columns.tolist(),
+        "data": schema_result.values.tolist()
+    }
 
 def tables_query(db_name):
     """
@@ -125,56 +165,93 @@ def meaning_query(db_name, table_name):
         table_name (str): The name of the table to query column meanings for. Must not be empty.
     
     Returns:
-        pandas.DataFrame: Query results containing column meanings and descriptions,
-                         or error message if parameters are missing or no meanings found.
+        dict: Dictionary containing query results with structure:
+              - Success: {"status": "success", "headers": [...], "data": [...]}
+              - Error: {"status": "error", "headers": ['message'], "data": [['Error message.']]}
     
     Example:
         >>> result = meaning_query("sales_db", "customers")
         >>> print(result)
-        # Returns DataFrame with explanations for each column in customers table
+        # Returns dict with explanations for each column in customers table
     """
     if not db_name or not table_name:
-        return pd.DataFrame({"message": ["請選擇資料庫和資料表"]})
+        return get_message("請選擇資料庫和資料表")
     
     result = get_meaning(db_name, table_name)
     
     if len(result) == 0:
-        return pd.DataFrame({"message": ["沒有找到相關欄位意義"]})
+        return get_message("Not Found.")
     
-    return pd.DataFrame(result)
+    return {"status": "success", "headers": result.columns.tolist(), "data": result.values.tolist()}
 
-def execute_sqlite(db_name: str, query: str):
+def meaning_search(db_name, keyword):
     """
-    Execute an SQL query on a specified SQLite database and return results.
-
-    This function allows direct SQL query execution on the selected database,
-    returning either the query results or error messages in a standardized format.
-
+    Search for column meanings using one or more keywords in a specific database.
+    
+    This function searches for columns whose names or meanings contain the specified
+    keyword(s), helping users find relevant fields across all tables in the database.
+    Multiple keywords can be separated by commas or spaces.
+    
     Args:
-        db_name (str): The name of the database to execute the query against. Must not be empty.
-        query (str): The SQL query to execute. Must not be empty.
-
+        db_name (str): The name of the database to search in. Must not be empty.
+        keyword (str): The search keyword(s) to match against column names or meanings.
+                      Multiple keywords can be separated by commas or spaces. Must not be empty.
+    
     Returns:
-        pandas.DataFrame: If query executes successfully, returns DataFrame with query results.
-                         If query fails or parameters are missing, returns error message in DataFrame.
-
+        dict: Dictionary containing query results with structure:
+              - Success: {"status": "success", "headers": [...], "data": [...]}
+              - Error: {"status": "error", "headers": ['message'], "data": [['Error message.']]}
+    
     Example:
-        >>> result = execute_sqlite("sales_db", "SELECT * FROM customers LIMIT 10")
+        >>> result = meaning_search("sales_db", "customer")
         >>> print(result)
-        # Returns DataFrame with first 10 rows from customers table
+        # Returns dict with all columns containing "customer" in their name or meaning
+        
+        >>> result = meaning_search("sales_db", "customer, order, product")
+        >>> print(result)
+        # Returns dict with all columns containing "customer", "order", or "product" in their name or meaning
     """
+    if not db_name or not keyword:
+        return get_message("Please input keyword.")
+    
+    # Split keywords by comma or space
+    keywords = [k.strip() for k in keyword.replace(' ', ',').split(',') if k.strip()]
+    
+    # Get results for each keyword and combine them
+    combined_results = []
+    for kw in keywords:
+        keyword_result = search_meaning(db_name, kw)
+        # Append non-empty results
+        if not keyword_result.empty:
+            combined_results.append(keyword_result)
+    
+    # Remove duplicates if any
+    # Convert DataFrames to a single DataFrame and remove duplicate rows
+    if combined_results:
+        result = pd.concat(combined_results, ignore_index=True).drop_duplicates()
+    else:
+        result = pd.DataFrame()
+    
+    if len(result) == 0:
+        return get_message("Not Found.")
+    
+    return {"status": "success", "headers": result.columns.tolist(), "data": result.values.tolist()}
+
+def execute_sqlite(db_name: str, query: str):
+
     if not db_name or not query:
         return pd.DataFrame({"message": ["請選擇資料庫和SQL"]})
     
     result = execute_sqlite_query(db_name, query)
     
-    if len(result) == 0:
-        return pd.DataFrame({"message": ["沒有找到相���資料"]})
+    # if len(result) == 0:
+    #     return pd.DataFrame({"message": ["沒有找到相關資料"]})
     
     if result['success']:
-        return pd.DataFrame(result['data'], columns=result['columns'])
+        # return pd.DataFrame(result['data'], columns=result['columns'])
+        return pd.DataFrame({"type":["OK"], "message": ["Success"]})
     else:
-        return pd.DataFrame({"message": [result['error']]})
+        return pd.DataFrame({"type":["Error"], "message": [result['error']]})
 
 
 # 建立 Gradio 界面
@@ -215,13 +292,13 @@ with gr.Blocks(title="資料庫查詢工具") as demo:
     with gr.Tab("欄位意義查詢"):
         with gr.Row():
             meaning_db = gr.Dropdown(choices=all_dbs, label="選擇資料庫", value=all_dbs[0] if all_dbs else None)
-            meaning_table = gr.Text(label="選擇資料表")
-        meaning_search = gr.Button("查詢欄位意義")
+            meaning_keyword = gr.Text(label="Keyword")
+        search_button = gr.Button("查詢欄位意義")
         meaning_result = gr.DataFrame(label="查詢結果")
         
         # 當資料庫選擇變更時，更新資料表下拉選單
         # meaning_db.change(update_tables, inputs=meaning_db, outputs=meaning_table)
-        meaning_search.click(meaning_query, inputs=[meaning_db, meaning_table], outputs=meaning_result)
+        search_button.click(meaning_search, inputs=[meaning_db, meaning_keyword], outputs=meaning_result)
 
     with gr.Tab("sqlite SQL 執行"):
         with gr.Row():
@@ -233,4 +310,4 @@ with gr.Blocks(title="資料庫查詢工具") as demo:
 
 # 啟動 Gradio 應用程式
 if __name__ == "__main__":
-    demo.launch(mcp_server=True, server_name="0.0.0.0",allowed_paths=["/"], share=True)
+    demo.launch(mcp_server=True, server_name="0.0.0.0", allowed_paths=["/"], share=True)

utils/tools.py

@@ -7,15 +7,15 @@ import shutil
 import uuid
 
 
-kb_df = pd.read_csv("./data/kb.csv")
+kb_df = pd.read_csv("./data/kb.csv", index_col=0)
 def get_kb(db_name, knowledge=None):
     if not knowledge:
         result = kb_df[(kb_df['db_name']==db_name)]
     else:
-        result = kb_df[(kb_df['db_name']==db_name) & (kb_df['knowledge'].str.contains(knowledge))]
+        result = kb_df[(kb_df['db_name']==db_name) & (kb_df['knowledge'].str.contains(knowledge, case=False))]
     return result
 
-schema_df = pd.read_csv("./data/db_schema.csv")
+schema_df = pd.read_csv("./data/db_schema.csv", index_col=0)
 def get_schema(db_name, table_name):
     result = schema_df[(schema_df['db_name']==db_name) & (schema_df['table_name']==table_name)]
     result = result[['schema', 'sample_data']]
@@ -27,10 +27,20 @@ def get_tables(db_name):
     tables = result['table_name'].to_list()
     return tables
 
-meaning_df = pd.read_csv("./data/column_meanings.csv")
+meaning_df = pd.read_csv("./data/column_meanings.csv", index_col=0)
 def get_meaning(db_name, table_name):
     result = meaning_df[(meaning_df['db_name']==db_name) & (meaning_df['table_name']==table_name)]
-    result = result[['column_name', 'meaning']]
+    result = result[['table_name', 'column_name', 'meaning']]
+    return result
+
+def search_meaning(db_name, keyword):
+    # Search in meaning column
+    result_meaning = meaning_df[(meaning_df['db_name']==db_name) & (meaning_df['meaning'].str.contains(keyword, case=False))]
+    # Search in column_name column
+    result_column = meaning_df[(meaning_df['db_name']==db_name) & (meaning_df['column_name'].str.contains(keyword, case=False))]
+    # Combine results and remove duplicates
+    result = pd.concat([result_meaning, result_column]).drop_duplicates()
+    result = result[['table_name', 'column_name', 'meaning']]
     return result
 
 def execute_sqlite_query(db_name, query):
@@ -67,12 +77,13 @@ def execute_sqlite_query(db_name, query):
         conn = sqlite3.connect(tmp_db_file)
         cursor = conn.cursor()
         
-        # 將查詢按分號分開並執行每條查詢
-        queries = query.split(';')
-        for q in queries:
-            q = q.strip()
-            if q:  # 確保不執行空查詢
-                cursor.execute(q)
+        # # 將查詢按分號分開並執行每條查詢
+        # queries = query.split(';')
+        # for q in queries:
+        #     q = q.strip()
+        #     if q:  # 確保不執行空查詢
+        #         cursor.execute(q)
+        cursor.executescript(query)
         
         # 獲取結果
         try: