BrandCultivationDataAnalysis/database/dao/mysql_dao.py

from database import  MySqlDatabaseHelper
import json
from database import CustConfig
import numpy as np

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._cust_table_name = "tads_brandcul_retail_cust_label"              # 商户信息表
        self._analysis_table_name = "tads_brandcul_analysis_index"             # 指标分析表
        
        self.cust_table_dao = self.CustTableDao(self.db_helper, self._cust_table_name)
        
        self._initialized = True
        
    class CustTableDao:
        """商户数据操作类"""
        def __init__(self, db_helper: MySqlDatabaseHelper, tablename):
            self.db_helper = db_helper
            self.tablename = tablename
            
        def load_data(self, features, corp_uuid):
            """读取商户数据"""
            features_column = ",".join(features)
            
            query = f"SELECT {features_column} FROM {self.tablename} WHERE corp_uuid = :corp_uuid"
            params = {"corp_uuid": corp_uuid}
            
            data = self.db_helper.load_data_with_page(query, params)
            
            return data
        
        def get_column_unique_value(self, column, corp_uuid):
            """获取指定列的唯一值列表"""
            query = f"SELECT {column} FROM {self.tablename} WHERE corp_uuid = :corp_uuid"
            params = {"corp_uuid": corp_uuid}
            
            data = self.db_helper.load_data_with_page(query, params)
            data = data[column].unique()
            
            # 将numpy array转换为Python列表，并确保所有元素都是Python原生类型
            if data is not None:
                # 处理numpy数据类型
                data = data.tolist()
                # 如果有numpy标量，转换为Python类型
                data = [self._convert_to_python_type(item) for item in data]
            
            return data
        
        def _convert_to_python_type(self, item):
            """将numpy/pandas类型转换为Python原生类型"""
            if isinstance(item, (np.integer, np.floating)):
                return item.item()
            elif isinstance(item, np.bool_):
                return bool(item)
            elif isinstance(item, np.ndarray):
                return item.tolist()
            else:
                return item
            
            
            
        
        
if __name__ == '__main__':
    features = ["busi_place_area_section"]
    
    dao = MySqlDao()
    corp_uuid = "00000000000000000000000011440101"
    
    features = CustConfig.FEATURES_COLUMNS
    features_countent = {}
    for feature in features:
        features_countent[feature] = dao.cust_table_dao.get_column_unique_value(feature, corp_uuid)
        
    json_str = json.dumps(features_countent, indent=4, ensure_ascii=False)
    
    print(json_str)
    with open('./cust_features_map.json', 'w', encoding='utf-8') as file:
        file.write(json_str)