BrandCultivation/utils/mock_data_to_database.py

#!/usr/bin/env python
# -*- encoding: utf-8 -*-
'''
@filename     : mock_data_to_database.py
@description     : 将mock数据写入到数据库中
@time     : 2025/01/31/00
@author     : Sherlock1011 & Min1027
@Version     : 1.0
'''
from dao.mysql_client import Mysql

import pandas as pd
from sqlalchemy import Column, Integer, VARCHAR, Float, DECIMAL
from sqlalchemy.ext.declarative import declarative_base

# 定义数据库表结构
Base = declarative_base()

class MockOrder(Base):
    __tablename__ = "mock_order"
    id = Column(Integer, primary_key=True, autoincrement=True)  # 添加主键列
    BB_RETAIL_CUSTOMER_CODE = Column(VARCHAR(50))
    PRODUCT_CODE = Column(VARCHAR(50))
    MONTH6_SALE_QTY = Column(DECIMAL(18, 6))
    MONTH6_SALE_AMT = Column(DECIMAL(18, 6))
    MONTH6_GROSS_PROFIT_RATE = Column(DECIMAL(18, 6))
    MONTH6_SALE_QTY_YOY = Column(DECIMAL(18, 6))
    MONTH6_SALE_QTY_MOM = Column(DECIMAL(18, 6))
    MONTH6_SALE_AMT_YOY = Column(DECIMAL(18, 6))
    MONTH6_SALE_AMT_MOM = Column(DECIMAL(18, 6))
    ORDER_FULLORDR_RATE = Column(DECIMAL(18, 6))
    CUSTOMER_REPURCHASE_RATE = Column(DECIMAL(18, 6))
    NEW_PRODUCT_ORDER_QTY_OCC = Column(DECIMAL(18, 6))
    LISTING_RATE = Column(DECIMAL(18, 6))
    OUT_STOCK_DAYS = Column(DECIMAL(18, 6))
    RETAIL_PRICE_INDEX = Column(DECIMAL(18, 6))
    
def insert_data(db, data_path):
    df = pd.read_excel(data_path)
    session = db.create_session()
    try:
        df.columns = ['BB_RETAIL_CUSTOMER_CODE', 
                      'PRODUCT_CODE', 
                      'MONTH6_SALE_QTY', 
                      'MONTH6_SALE_AMT', 
                      'MONTH6_GROSS_PROFIT_RATE',
                      'MONTH6_SALE_QTY_YOY', 
                      'MONTH6_SALE_QTY_MOM', 
                      'MONTH6_SALE_AMT_YOY', 
                      'MONTH6_SALE_AMT_MOM', 
                      'ORDER_FULLORDR_RATE',
                      'CUSTOMER_REPURCHASE_RATE', 
                      'NEW_PRODUCT_ORDER_QTY_OCC', 
                      'LISTING_RATE', 
                      'OUT_STOCK_DAYS', 
                      'RETAIL_PRICE_INDEX',
                      ]  # 确保列名匹配
        session.bulk_insert_mappings(MockOrder, df.to_dict(orient='records'))
        session.commit()
        print("数据成功插入数据库")
        
    except Exception as e:
        session.rollback()
        print(f"插入数据时出错: {e}")
        
    finally:
        session.close()
    
        
if __name__ == "__main__":
    data_path = "./data/order.xlsx"
    # 创建数据库链接
    db = Mysql()
    

    # 创建表（如果不存在）
    Base.metadata.create_all(db.engine)
    insert_data(db, data_path)
    
    db.closed()