BrandCultivation/database/dao/mysql_dao.py

from database import MySqlDatabaseHelper
from sqlalchemy import text
import pandas as pd

class MySqlDao:
    _instance = None
    
    def __new__(cls):
        if not cls._instance:
            cls._instance = super(MySqlDao, cls).__new__(cls)
            cls._instance._initialized = False
        return cls._instance
    
    
    def __init__(self):
        if self._initialized:
            return
        
        self.db_helper = MySqlDatabaseHelper()
        self._product_tablename = "tads_brandcul_product_info_f"
        self._cust_tablename = "tads_brandcul_cust_info_f"
        self._order_tablename = "tads_brandcul_consumer_order"
        self._eval_order_name = "tads_brandcul_consumer_order_check"
        self._mock_order_tablename = "yunfu_mock_data"
        self._shopping_tablename = "tads_brandcul_cust_info_lbs_f"
        # self._shopping_tablename = "yunfu_shopping_mock_data"
        
        self._initialized = True
        
    def load_product_data(self, city_uuid):
        """从数据库中读取商品信息"""
        query = f"SELECT * FROM {self._product_tablename} WHERE city_uuid = :city_uuid"
        params = {"city_uuid": city_uuid}
        
        data = self.db_helper.load_data_with_page(query, params)
        return data
        
    def load_cust_data(self, city_uuid):
        """从数据库中读取商户信息"""
        query = f"SELECT * FROM {self._cust_tablename} WHERE BA_CITY_ORG_CODE = :city_uuid"
        params = {"city_uuid": city_uuid}
        
        data = self.db_helper.load_data_with_page(query, params)
        return data
    
    def load_order_data(self, city_uuid):
        """从数据库中读取订单信息"""
        query = f"SELECT * FROM {self._order_tablename} WHERE city_uuid = :city_uuid"
        params = {"city_uuid": city_uuid}
        
        data = self.db_helper.load_data_with_page(query, params)
        data.drop('stat_month', axis=1, inplace=True)
        data.drop('city_uuid', axis=1, inplace=True)
        
        return data
    
    def load_eval_order_data(self, city_uuid):
        """从数据库中读取订单信息"""
        query = f"SELECT * FROM {self._eval_order_name} WHERE city_uuid = :city_uuid"
        params = {"city_uuid": city_uuid}
        
        data = self.db_helper.load_data_with_page(query, params)
        data.drop('stat_month', axis=1, inplace=True)
        data.drop('city_uuid', axis=1, inplace=True)
        
        return data
    
    def load_mock_order_data(self):
        """从数据库中读取mock的订单信息"""
        query = f"SELECT * FROM {self._mock_order_tablename}"
        
        data = self.db_helper.load_data_with_page(query, {})
        
        return data
    
    def load_shopping_data(self, city_uuid):
        """从数据库中读取商圈数据"""
        query = f"SELECT * FROM {self._shopping_tablename} WHERE city_uuid = :city_uuid"
        params = {"city_uuid": city_uuid}
        
        data = self.db_helper.load_data_with_page(query, params)
        
        return data
    
    def get_cust_list(self, city_uuid):
        """获取商户列表"""
        data = self.load_cust_data(city_uuid)
        cust_list = data["BB_RETAIL_CUSTOMER_CODE"].to_list()
        if len(cust_list) == 0:
            return []
        
        return cust_list
    
    def get_product_by_id(self, city_uuid, product_id):
        """根据city_uuid 和 product_id 从表中获取拼柜信息"""
        query = f"""
            SELECT *
            FROM {self._product_tablename}
            WHERE city_uuid = :city_uuid
            AND product_code = :product_id
        """
        params = {"city_uuid": city_uuid, "product_id": product_id}
        data = self.db_helper.load_data_with_page(query, params)
        
        return data
    
    def get_cust_by_ids(self, city_uuid, cust_id_list):
        """根据零售户列表查询其信息"""
        if not cust_id_list:
            return None
        
        cust_id_str = ",".join([f"'{cust_id}'" for cust_id in cust_id_list])
        query = f"""
            SELECT *
            FROM {self._cust_tablename}
            WHERE BA_CITY_ORG_CODE = :city_uuid
            AND BB_RETAIL_CUSTOMER_CODE IN ({cust_id_str})
        """
        params = {"city_uuid": city_uuid}
        data = self.db_helper.load_data_with_page(query, params)
        
        return data
    
    def get_shop_by_ids(self, city_uuid, cust_id_list):
        """根据零售户列表查询其信息"""
        if not cust_id_list:
            return None
        
        cust_id_str = ",".join([f"'{cust_id}'" for cust_id in cust_id_list])
        query = f"""
            SELECT *
            FROM {self._shopping_tablename}
            WHERE city_uuid = :city_uuid
            AND cust_code IN ({cust_id_str})
        """
        params = {"city_uuid": city_uuid}
        data = self.db_helper.load_data_with_page(query, params)
        
        return data
    
    def get_product_by_ids(self, city_uuid, product_id_list):
        """根据product_code列表查询其信息"""
        if not product_id_list:
            return None
        
        product_id_str = ",".join([f"'{product_id}'" for product_id in product_id_list])
        query = f"""
            SELECT *
            FROM {self._product_tablename}
            WHERE city_uuid = :city_uuid
            AND product_code IN ({product_id_str})
        """
        params = {"city_uuid": city_uuid}
        data = self.db_helper.load_data_with_page(query, params)
        
        return data
    
    def get_order_by_product_ids(self, city_uuid, product_ids):
        """获取指定香烟列表的所有售卖记录"""
        if not product_ids:
            return None
        
        product_ids_str = ",".join([f"'{product_code}'" for product_code in product_ids])
        query = f"""
            SELECT *
            FROM {self._order_tablename}
            WHERE city_uuid = :city_uuid
            AND product_code IN ({product_ids_str})
        """
        params = {"city_uuid": city_uuid}
        data = self.db_helper.load_data_with_page(query, params)
        
        return data
    
    def get_order_by_product(self, city_uuid, product_id):
        query = f"""
            SELECT *
            FROM {self._order_tablename}
            WHERE city_uuid = :city_uuid
            AND product_code = :product_id
        """
        params = {"city_uuid": city_uuid, "product_id": product_id}
        data = self.db_helper.load_data_with_page(query, params)
        
        return data
    
    def get_eval_order_by_product(self, city_uuid, product_id):
        query = f"""
            SELECT *
            FROM {self._eval_order_name}
            WHERE city_uuid = :city_uuid
            AND product_code = :product_id
        """
        params = {"city_uuid": city_uuid, "product_id": product_id}
        data = self.db_helper.load_data_with_page(query, params)
        
        return data
    
    def get_order_by_cust(self, city_uuid, cust_id):
        query = f"""
            SELECT *
            FROM {self._order_tablename}
            WHERE city_uuid = :city_uuid
            AND cust_code = :cust_id
        """
        params = {"city_uuid": city_uuid, "cust_id": cust_id}
        data = self.db_helper.load_data_with_page(query, params)
        
        return data
    
    def get_order_by_cust_and_product(self, city_uuid, cust_id, product_id):
        query = f"""
            SELECT *
            FROM {self._order_tablename}
            WHERE city_uuid = :city_uuid
            AND cust_code = :cust_id
            AND product_code =:product_id
        """
        params = {"city_uuid": city_uuid, "cust_id": cust_id, "product_id": product_id}
        data = self.db_helper.load_data_with_page(query, params)
        
        return data
    
    def get_product_from_order(self, city_uuid):
        query = f"SELECT DISTINCT product_code FROM {self._order_tablename} WHERE city_uuid = :city_uuid"
        params = {"city_uuid": city_uuid}
        
        data = pd.DataFrame(self.db_helper.fetch_all(text(query), params))
        
        return data
    
    def data_preprocess(self, data: pd.DataFrame):
        
        data.drop(["cust_uuid", "longitude", "latitude", "range_radius"], axis=1, inplace=True)
        remaining_cols = data.columns.drop(["city_uuid", "cust_code"])
        col_with_missing = remaining_cols[data[remaining_cols].isnull().any()].tolist() # 判断有缺失的字段
        col_all_missing = remaining_cols[data[remaining_cols].isnull().all()].to_list() # 全部缺失的字段
        col_partial_missing = list(set(col_with_missing) - set(col_all_missing)) # 部分缺失的字段
        
        for col in col_partial_missing:
            data[col] = data[col].fillna(data[col].mean())
        
        for col in col_all_missing:
            data[col] = data[col].fillna(0).infer_objects(copy=False)
        
        
if __name__ == "__main__":
    dao = MySqlDao()
    city_uuid = "00000000000000000000000011445301"
    
    order_data = dao.load_order_data(city_uuid)
    order_data["sale_qty"] = order_data["sale_qty"].fillna(0)
    order_data = order_data.infer_objects(copy=False)
        
        # 将销售量进行分组求和
    order_data = order_data.groupby(["cust_code", "product_code"], as_index=False)["sale_qty"].sum()
    
    cust_data = dao.load_cust_data(city_uuid)
    cust_data = cust_data["BB_RETAIL_CUSTOMER_NAME"]
    
    sale_data = order_data.merge(cust_data, left_on='cust_code', right_on='BB_RETAIL_CUSTOMER_CODE', how="inner")
    
    sale_data.to_csv("./data/sale.csv", index=False)