122 lines
3.9 KiB
Python
Executable File
122 lines
3.9 KiB
Python
Executable File
# 1.导入必要的库和模块
|
|
from flask import Flask, jsonify, Blueprint
|
|
import pymysql
|
|
from flask_cors import CORS
|
|
from contextlib import contextmanager
|
|
from flask.json import JSONEncoder
|
|
from decimal import Decimal
|
|
|
|
|
|
# 2.创建 Flask 应用并配置
|
|
app = Flask(__name__)
|
|
app.config['JSON_AS_ASCII'] = False # 重要，让 JSON 数据按原字符编码输出，不转成 ASCII
|
|
app.config['JSONIFY_MIMETYPE'] = 'application/json; charset=utf-8'
|
|
CORS(app) # 解决跨域问题
|
|
|
|
class CustomJSONEncoder(JSONEncoder):
|
|
def default(self, obj):
|
|
if isinstance(obj, Decimal):
|
|
return float(obj) # 或 str(obj)
|
|
return super().default(obj)
|
|
|
|
app.json_encoder = CustomJSONEncoder
|
|
# 3.数据库连接上下文管理器
|
|
@contextmanager
|
|
def get_db_connection():
|
|
try:
|
|
connection = pymysql.connect(
|
|
host='hadoop102',
|
|
user='root',
|
|
port=3306,
|
|
password='123456',
|
|
database='marketing',
|
|
charset='utf8mb4',
|
|
cursorclass=pymysql.cursors.DictCursor
|
|
)
|
|
yield connection
|
|
except Exception as e:
|
|
print(f"Error: {e}")
|
|
finally:
|
|
if connection:
|
|
connection.close()
|
|
|
|
# 4.创建蓝图
|
|
data_bp = Blueprint('data', __name__)
|
|
|
|
# 5.定义API路由
|
|
# sales_month的数据接口
|
|
@data_bp.route('/api/sales_month', methods=['GET'])
|
|
def sales_month():
|
|
with get_db_connection() as connection:
|
|
with connection.cursor() as cursor:
|
|
sql = "SELECT * FROM sales_month"
|
|
cursor.execute(sql)
|
|
results = cursor.fetchall()
|
|
return jsonify(results)
|
|
|
|
# sales_region的数据接口
|
|
@data_bp.route('/api/sales_region', methods=['GET'])
|
|
def sales_region():
|
|
with get_db_connection() as connection:
|
|
with connection.cursor() as cursor:
|
|
sql = "SELECT * FROM sales_region"
|
|
cursor.execute(sql)
|
|
results = cursor.fetchall()
|
|
return jsonify(results)
|
|
|
|
# sales_manager的数据接口
|
|
@data_bp.route('/api/sales_manager', methods=['GET'])
|
|
def sales_manager():
|
|
with get_db_connection() as connection:
|
|
with connection.cursor() as cursor:
|
|
sql = "SELECT * FROM sales_manager"
|
|
cursor.execute(sql)
|
|
results = cursor.fetchall()
|
|
return jsonify(results)
|
|
|
|
# sales_province的数据接口
|
|
@data_bp.route('/api/sales_province', methods=['GET'])
|
|
def sales_province():
|
|
with get_db_connection() as connection:
|
|
with connection.cursor() as cursor:
|
|
sql = "SELECT * FROM sales_province"
|
|
cursor.execute(sql)
|
|
results = cursor.fetchall()
|
|
return jsonify(results)
|
|
|
|
# sales_product的数据接口
|
|
@data_bp.route('/api/sales_product', methods=['GET'])
|
|
def combine_data():
|
|
result = []
|
|
with get_db_connection() as conn:
|
|
cursor = conn.cursor()
|
|
# 直接使用 JOIN 语句查询关联数据
|
|
sql = """
|
|
SELECT mc.name AS main_name, mc.value, sc.name AS sub_name, sc.value
|
|
FROM main_categories mc
|
|
JOIN sub_categories sc ON mc.id = sc.main_category_id
|
|
"""
|
|
cursor.execute(sql)
|
|
rows = cursor.fetchall()
|
|
category_dict = {}
|
|
for row in rows:
|
|
main_name, main_value, sub_name, sub_value = row.values()
|
|
if main_name not in category_dict:
|
|
category_dict[main_name] = {
|
|
"name": main_name,
|
|
"value": main_value,
|
|
"children": []
|
|
}
|
|
category_dict[main_name]["children"].append({
|
|
"name": sub_name,
|
|
"value": sub_value
|
|
})
|
|
result = list(category_dict.values())
|
|
|
|
return jsonify(result)
|
|
|
|
# 6.注册蓝图并运行应用
|
|
app.register_blueprint(data_bp)
|
|
|
|
if __name__ == '__main__':
|
|
app.run(debug=True) |