class SchemaManager:
    def __init__(self, db_config):
        self.db_config = db_config
        self.connection = None
        self.schema_metadata = None
    
    def fetch_metadata(self):
        # 获取数据库表结构信息
        pass
    
    def build_vector_index(self):
        # 将表结构向量化用于快速检索
        pass

import dashscope
from dashscope import Generation
import json
import os

# 设置通义千问API密钥
dashscope.api_key = os.environ.get('DASHSCOPE_API_KEY')

# 定义可用的函数
functions = [
    {
        "name": "execute_sql_query",
        "description": "执行SQL查询并返回结果",
        "parameters": {
            "type": "object",
            "properties": {
                "sql": {
                    "type": "string",
                    "description": "要执行的SQL查询语句"
                }
            },
            "required": ["sql"]
        }
    }
]

# 模拟的数据库表结构信息
database_schema = """
数据库中有以下表:
1. 用户表(users): 包含id, name, email, city字段
2. 订单表(orders): 包含id, user_id, product, amount, order_date字段
3. 产品表(products): 包含id, name, price, category字段
"""

def text_to_sql(user_query):
    """将自然语言转换为SQL并执行"""
    
    try:
        # 调用通义千问Qwen-Plus模型，支持Function Calling
        response = Generation.call(
            model='qwen-plus',
            messages=[
                {
                    "role": "system", 
                    "content": f"你是一个SQL专家。请根据以下数据库结构生成SQL查询:\n{database_schema}"
                },
                {
                    "role": "user", 
                    "content": user_query
                }
            ],
            functions=functions,
            result_format='message'
        )

# 解析模型响应
        message = response.output.choices[0].message
        
        # 检查是否要求调用函数
        if hasattr(message, 'function_call') and message.function_call:
            function_name = message.function_call.name
            function_args = json.loads(message.function_call.arguments)
            
            # 执行相应的函数
            if function_name == "execute_sql_query":
                return execute_sql_query(function_args["sql"])

else:
            # 如果没有函数调用，尝试直接提取SQL
            sql = extract_sql_from_text(message.content)
            if sql:
                return execute_sql_query(sql, "从模型响应中提取SQL")
            else:
                return "未能生成合适的SQL查询"
    
    except Exception as e:
        return f"API调用失败: {str(e)}"

def extract_sql_from_text(text):
    """从文本中提取SQL语句"""
    import re
    
    # 查找SELECT语句
    select_pattern = r"SELECT\s+.*?FROM\s+.*?(?:WHERE\s+.*?)?(?:ORDER BY\s+.*?)?(?:LIMIT\s+.*?)?;?"
    matches = re.findall(select_pattern, text, re.IGNORECASE | re.DOTALL)
    
    if matches:
        return matches[0].strip()
    
    # 查找其他类型的SQL语句
    other_patterns = [
        r"INSERT INTO\s+.*?VALUES\s*\(.*?\);?",
        r"UPDATE\s+.*?SET\s+.*?WHERE\s+.*?;?",
        r"DELETE FROM\s+.*?WHERE\s+.*?;?"
    ]
    
    for pattern in other_patterns:
        matches = re.findall(pattern, text, re.IGNORECASE | re.DOTALL)
        if matches:
            return matches[0].strip()
    
    return None

def execute_sql_query(sql, thought=""):
    """执行SQL查询的函数"""
    # 在实际应用中，这里会连接真实数据库执行查询
    # 这里只是返回模拟结果
    
    print(f"思考过程: {thought}")
    print(f"执行的SQL: {sql}")
    
    # 模拟不同查询的结果
    sql_lower = sql.lower()
    
    if "select name, city from users" in sql_lower:
        return [
            {"name": "张三", "city": "北京"},
            {"name": "李四", "city": "上海"},
            {"name": "王五", "city": "广州"}
        ]
    # ... 其他条件判断

# 测试示例
if __name__ == "__main__":
    # 示例查询
    queries = [
        "显示所有用户的姓名和所在城市",
        "查询订单中的产品和数量",
        "删除所有测试用户",
        "显示所有产品名称和价格",
        "查询北京的用户"
    ]
    
    for query in queries:
        print(f"\n{'='*50}")
        print(f"用户查询: {query}")
        result = text_to_sql(query)
        print(f"查询结果: {result}")
        print(f"{'='*50}")

import os
import json
import pymysql
import dashscope
from dashscope import Generation
# 设置通义千问API密钥
dashscope.api_key = os.environ.get('DASHSCOPE_API_KEY')

# 数据库配置
    db_config = {
        "host": "localhost",
        "user": "your_username",
        "password": "your_password",
        "database": "your_database",
        "charset": "utf8mb4"
    }
    
    # 初始化Text2SQL系统
    text2sql = SimpleText2SQL(db_config)

def get_database_schema(self):
    """获取数据库表结构信息"""
    try:
        connection = pymysql.connect(**self.db_config)
        schema = {}
        
        with connection.cursor() as cursor:
            # 获取所有表名
            cursor.execute("SHOW TABLES")
            tables = cursor.fetchall()
            
            for table in tables:
                table_name = table[0]
                schema[table_name] = {"columns": []}
                
                # 获取表的列信息
                cursor.execute(f"DESCRIBE {table_name}")
                columns = cursor.fetchall()

def generate_sql(self, user_query):
    """使用Qwen生成SQL查询"""
    if not self.schema_info:
        self.get_database_schema()
    
    # 构建schema描述
    schema_desc = "数据库表结构:\n"
    for table_name, table_info in self.schema_info.items():
        schema_desc += f"- {table_name}: "
        schema_desc += ", ".join([f"{col['name']}({col['type']})" for col in table_info["columns"]])
        schema_desc += "\n"
    
    # 构建提示词
    prompt = f"""你是一个SQL专家。请根据以下数据库结构，将用户的自然语言查询转换为MySQL SQL语句。
{schema_desc}
请只输出SQL语句，不要输出其他任何内容。
用户查询: {user_query}
SQL语句:"""

# 调用通义千问API
response = Generation.call(
    model='qwen-turbo',  # 使用qwen-turbo以节省成本
    messages=[{'role': 'user', 'content': prompt}],
    temperature=0.1
)
# 提取SQL语句
sql = response.output.choices[0].message.content.strip()

def execute_query(self, sql):
    """执行SQL查询并返回结果"""
    # 安全检查：只允许SELECT查询
    if not sql.lower().strip().startswith('select'):
        return {"error": "只允许执行SELECT查询"}
    
    try:
        connection = pymysql.connect(**self.db_config)
        
        with connection.cursor(pymysql.cursors.DictCursor) as cursor:
            cursor.execute(sql)
            results = cursor.fetchall()
            return results

import dashscope
from dashscope import Generation
import json
import os
# 设置通义千问API密钥
dashscope.api_key = os.environ.get('DASHSCOPE_API_KEY')
# 定义可用的函数
functions = [
    {
        "name": "execute_sql_query",
        "description": "执行SQL查询并返回结果",
        "parameters": {
            "type": "object",
            "properties": {
                "sql": {
                    "type": "string",
                    "description": "要执行的SQL查询语句"
                }
            },
            "required": ["sql"]
        }
    }
]
# 模拟的数据库表结构信息
database_schema = """
数据库中有以下表:
1. 用户表(users): 包含id, name, email, city字段
2. 订单表(orders): 包含id, user_id, product, amount, order_date字段
3. 产品表(products): 包含id, name, price, category字段
"""
def text_to_sql(user_query):
    """将自然语言转换为SQL并执行"""
    
    try:
        # 调用通义千问Qwen-Plus模型，支持Function Calling
        response = Generation.call(
            model='qwen-plus',
            messages=[
                {
                    "role": "system", 
                    "content": f"你是一个SQL专家。请根据以下数据库结构生成SQL查询:\n{database_schema}"
                },
                {
                    "role": "user", 
                    "content": user_query
                }
            ],
            functions=functions,
            result_format='message'
        )
        
        # 解析模型响应
        message = response.output.choices[0].message
        print(message)
        # 检查是否要求调用函数
        if hasattr(message, 'function_call') and message.function_call:
            # 检查function_call是对象还是字典
            if isinstance(message.function_call, dict):
                function_name = message.function_call.get('name')
                function_args = json.loads(message.function_call.get('arguments', '{}'))
            else:
                try:
                    function_name = message.function_call.name
                    function_args = json.loads(message.function_call.arguments)
                except AttributeError:
                    # 如果访问属性失败，尝试使用字典方式访问
                    function_name = message.function_call.get('name')
                    function_args = json.loads(message.function_call.get('arguments', '{}'))
            
            # 执行相应的函数
            if function_name == "execute_sql_query":
                return execute_sql_query(function_args["sql"])
        else:
            # 如果没有函数调用，尝试直接提取SQL
            sql = extract_sql_from_text(message.content)
            if sql:
                return execute_sql_query(sql, "从模型响应中提取SQL")
            else:
                return "未能生成合适的SQL查询"
    
    except Exception as e:
        return f"API调用失败: {str(e)}"
def extract_sql_from_text(text):
    """从文本中提取SQL语句"""
    import re
    
    # 查找SELECT语句
    select_pattern = r"SELECT\s+.*?FROM\s+.*?(?:WHERE\s+.*?)?(?:ORDER BY\s+.*?)?(?:LIMIT\s+.*?)?;?"
    matches = re.findall(select_pattern, text, re.IGNORECASE | re.DOTALL)
    
    if matches:
        return matches[0].strip()
    
    # 查找其他类型的SQL语句
    other_patterns = [
        r"INSERT INTO\s+.*?VALUES\s*\(.*?\);?",
        r"UPDATE\s+.*?SET\s+.*?WHERE\s+.*?;?",
        r"DELETE FROM\s+.*?WHERE\s+.*?;?"
    ]
    
    for pattern in other_patterns:
        matches = re.findall(pattern, text, re.IGNORECASE | re.DOTALL)
        if matches:
            return matches[0].strip()
    
    return None
def execute_sql_query(sql, thought=""):
    """执行SQL查询的函数"""
    # 在实际应用中，这里会连接真实数据库执行查询
    # 这里只是返回模拟结果
    
    print(f"思考过程: {thought}")
    print(f"执行的SQL: {sql}")
    
    # 模拟不同查询的结果
    sql_lower = sql.lower()
    
    if "select name, city from users" in sql_lower:
        return [
            {"name": "张三", "city": "北京"},
            {"name": "李四", "city": "上海"},
            {"name": "王五", "city": "广州"}
        ]
    elif "select product, amount from orders" in sql_lower:
        return [
            {"product": "笔记本电脑", "amount": 2},
            {"product": "智能手机", "amount": 5},
            {"product": "平板电脑", "amount": 3}
        ]
    elif "select * from users" in sql_lower:
        return [
            {"id": 1, "name": "张三", "email": "zhangsan@example.com", "city": "北京"},
            {"id": 2, "name": "李四", "email": "lisi@example.com", "city": "上海"},
            {"id": 3, "name": "王五", "email": "wangwu@example.com", "city": "广州"}
        ]
    elif "select name, price from products" in sql_lower:
        return [
            {"name": "笔记本电脑", "price": 5999.00},
            {"name": "智能手机", "price": 3999.00},
            {"name": "平板电脑", "price": 2999.00}
        ]
    elif "delete" in sql_lower:
        return [{"message": "删除操作已执行", "rows_affected": 5}]
    elif "update" in sql_lower:
        return [{"message": "更新操作已执行", "rows_affected": 3}]
    elif "insert" in sql_lower:
        return [{"message": "插入操作已执行", "rows_affected": 1}]
    else:
        return [{"message": "查询执行成功", "rows_affected": 10}]
# 测试示例
if __name__ == "__main__":
    # 示例查询
    queries = [
        "显示所有用户的姓名和所在城市",
        "查询订单中的产品和数量",
        "删除所有测试用户",
        "显示所有产品名称和价格",
        "查询北京的用户"
    ]
    
    for query in queries:
        print(f"\n{'='*50}")
        print(f"用户查询: {query}")
        result = text_to_sql(query)
        print(f"查询结果: {result}")
        print(f"{'='*50}")

import os
import json
import pymysql
import dashscope
from dashscope import Generation
# 设置通义千问API密钥
dashscope.api_key = os.environ.get('DASHSCOPE_API_KEY')
class SimpleText2SQL:
    def __init__(self, db_config):
        self.db_config = db_config
        self.schema_info = None
    
    def get_database_schema(self):
        """获取数据库表结构信息"""
        try:
            connection = pymysql.connect(**self.db_config)
            schema = {}
            
            with connection.cursor() as cursor:
                # 获取所有表名
                cursor.execute("SHOW TABLES")
                tables = cursor.fetchall()
                
                for table in tables:
                    table_name = table[0]
                    schema[table_name] = {"columns": []}
                    
                    # 获取表的列信息
                    cursor.execute(f"DESCRIBE {table_name}")
                    columns = cursor.fetchall()
                    
                    for col in columns:
                        schema[table_name]["columns"].append({
                            "name": col[0],
                            "type": col[1],
                            "nullable": col[2],
                            "key": col[3],
                            "default": col[4],
                            "extra": col[5]
                        })
            
            self.schema_info = schema
            return schema
            
        except Exception as e:
            print(f"获取数据库结构失败: {str(e)}")
            return None
        finally:
            if connection:
                connection.close()
    
    def generate_sql(self, user_query):
        """使用Qwen生成SQL查询"""
        if not self.schema_info:
            self.get_database_schema()
        
        # 构建schema描述
        schema_desc = "数据库表结构:\n"
        for table_name, table_info in self.schema_info.items():
            schema_desc += f"- {table_name}: "
            schema_desc += ", ".join([f"{col['name']}({col['type']})" for col in table_info["columns"]])
            schema_desc += "\n"
        
        # 构建提示词
        prompt = f"""你是一个SQL专家。请根据以下数据库结构，将用户的自然语言查询转换为MySQL SQL语句。
{schema_desc}
请只输出SQL语句，不要输出其他任何内容。
用户查询: {user_query}
SQL语句:"""
        
        try:
            # 调用通义千问API
            response = Generation.call(
                model='qwen-turbo',  # 使用qwen-turbo以节省成本
                messages=[{'role': 'user', 'content': prompt}],
                temperature=0.1
            )
            print(f"Qwen API响应: {response}")
            
            # 检查API响应是否有效 or not response.output.choices
            if not response or not hasattr(response, 'output') :
                print(f"API返回无效响应: {response}")
                return None
            
            try:
                # 提取SQL语句
                # sql = response.output.choices[0].message.content.strip()
                sql = response.output.text.strip()
                
                # 清理SQL语句（移除可能的代码标记）
                if sql.startswith('```sql'):
                    sql = sql[6:]
                if sql.startswith('```'):
                    sql = sql[3:]
                if sql.endswith('```'):
                    sql = sql[:-3]
            except Exception as e:
                print(f"解析API响应失败: {str(e)}")
                return None
            
            return sql.strip()
            
        except Exception as e:
            print(f"生成SQL失败: {str(e)}")
            return None
    
    def execute_query(self, sql):
        """执行SQL查询并返回结果"""
        # 安全检查：只允许SELECT查询
        if not sql.lower().strip().startswith('select'):
            return {"error": "只允许执行SELECT查询"}
        
        try:
            connection = pymysql.connect(**self.db_config)
            
            with connection.cursor(pymysql.cursors.DictCursor) as cursor:
                cursor.execute(sql)
                results = cursor.fetchall()
                return results
                
        except Exception as e:
            return {"error": f"执行查询失败: {str(e)}"}
        finally:
            if connection:
                connection.close()
    
    def query(self, user_query):
        """处理用户查询的完整流程"""
        print(f"用户查询: {user_query}")
        
        # 生成SQL
        sql = self.generate_sql(user_query)
        if not sql:
            return {"error": "无法生成SQL语句"}
        
        print(f"生成的SQL: {sql}")
        
        # 执行查询
        result = self.execute_query(sql)
        return result
# 使用示例
if __name__ == "__main__":
    # 数据库配置
    db_config = {
        "host": "localhost",
        "user": "root",
        "password": "Aa123456!",
        "database": "ecommerce_db",
        "charset": "utf8mb4"
    }
    
    # 初始化Text2SQL系统
    text2sql = SimpleText2SQL(db_config)
    
    # 示例查询
    queries = [
        "显示所有用户",
        "查询订单数量最多的前5个产品",
        "查找最近一个月注册的用户",
        "统计每个城市的用户数量"
    ]
    
    for query in queries:
        print("\n" + "="*50)
        result = text2sql.query(query)
        
        if "error" in result:
            print(f"错误: {result['error']}")
        else:
            print(f"查询结果 (显示前5行):")
            for i, row in enumerate(result[:5]):
                print(f"  {i+1}. {row}")
        
        print("="*50)