大模型即服务

借助大模型语义化直接操作数据库

除了效率有点低，有时候有点傻，有点费token

AI生成

api.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
from flask import Flask, request, jsonify
import pymysql
import openpyxl
from qcloud_cos import CosConfig, CosS3Client
import tempfile
import os

app = Flask(__name__)

# 数据库连接配置
db_config = {
    'host': 'xx',
    'user': 'xx',
    'password': 'xx',
    'database': 'xx'
}

# 腾讯云COS配置
cos_config = {
    'secret_id': 'xx',
    'secret_key': 'xx',
    'region': 'xx',
    'bucket': 'xx'
}

def execute_sql(operation, sql):
    connection = pymysql.connect(**db_config)
    try:
        with connection.cursor() as cursor:
            if operation == '查':
                cursor.execute(sql)
                result = cursor.fetchall()
                return result
            elif operation in ['增', '删', '改']:
                cursor.execute(sql)
                connection.commit()
                return {'status': 'success'}
            else:
                return {'error': '无效的操作名'}
    except Exception as e:
        return {'error': str(e)}
    finally:
        connection.close()

def save_to_excel(data):
    wb = openpyxl.Workbook()
    ws = wb.active
    for row in data:
        ws.append(row)
    temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.xlsx')
    wb.save(temp_file.name)
    return temp_file.name

def upload_to_cos(file_path):
    config = CosConfig(Region=cos_config['region'], SecretId=cos_config['secret_id'], SecretKey=cos_config['secret_key'])
    client = CosS3Client(config)
    file_name = os.path.basename(file_path)
    response = client.upload_file(
        Bucket=cos_config['bucket'],
        LocalFilePath=file_path,
        Key=file_name
    )
    url = f"https://{cos_config['bucket']}.cos.{cos_config['region']}.myqcloud.com/{file_name}"
    return url

def execute_sql_with_headers(sql):
    connection = pymysql.connect(**db_config)
    try:
        with connection.cursor() as cursor:
            cursor.execute(sql)
            result = cursor.fetchall()
            # 获取列名或别名
            headers = [desc[0] for desc in cursor.description]
            return headers, result
    except Exception as e:
        return None, {'error': str(e)}
    finally:
        connection.close()

def save_to_excel_with_headers(headers, data):
    wb = openpyxl.Workbook()
    ws = wb.active
    ws.append(headers)  # 添加表头
    for row in data:
        ws.append(row)
    temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.xlsx')
    wb.save(temp_file.name)
    return temp_file.name

@app.route('/export', methods=['POST'])
def export():
    data = request.json
    sql = data.get('sql')
    headers, query_result = execute_sql_with_headers(sql)
    if 'error' in query_result:
        return jsonify(query_result)
    
    file_path = save_to_excel_with_headers(headers, query_result)
    download_url = upload_to_cos(file_path)
    os.remove(file_path)  # 删除临时文件
    return jsonify({'download_url': download_url})

@app.route('/tables', methods=['GET'])
def get_tables():
    connection = pymysql.connect(**db_config)
    try:
        with connection.cursor() as cursor:
            # 获取所有表名
            cursor.execute("SHOW TABLES")
            tables = cursor.fetchall()
            
            # 获取每个表的创建语句
            table_structures = []
            for (table_name,) in tables:
                cursor.execute(f"SHOW CREATE TABLE {table_name}")
                _, create_statement = cursor.fetchone()
                table_structures.append({
                    "表名": table_name,
                    "结构": create_statement
                })
            
            return jsonify(table_structures)
    except Exception as e:
        return jsonify({'error': str(e)})
    finally:
        connection.close()

@app.route('/execute', methods=['POST'])
def execute():
    data = request.json
    sql = data.get('sql')
    operation = data.get('operation')
    result = execute_sql(operation, sql)
    return jsonify(result)

if __name__ == '__main__':
    app.run(debug=True, port=12345)