data_analysis_wind_turbine/dataStorage_datang/info_turbine_data.py

import pandas as pd
import pymysql
from sqlalchemy import create_engine, text
import warnings
from typing import Optional, Dict, Any, List
import logging

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# 映射中文表头到英文列名
HEADER_MAPPING = {
    '风机id': 'turbine_id',
    '风场id': 'wind_farm_id',
    '风机名称': 'turbine_name',
    '风机英文名称': 'turbine_name_en',
    'USCADA别名': 'uscada_alias',
    '海拔': 'altitude',
    '纬度': 'latitude',
    '经度': 'longitude',
    '主控接口版本': 'main_control_version',
    '所属馈线ID': 'feeder_id',
    '所属期ID': 'phase_id',
    '轮毂高度': 'hub_height',
    '指标统计开始日期': 'stat_start_date',
    '投产运行日期': 'operation_date',
    '合同功率曲线（标准空气密度）': 'contract_power_curve_std',
    '合同功率曲线（当地空气密度）': 'contract_power_curve_local',
    '电量倍率': 'power_multiplier',
    '电量跳变斜率': 'power_jump_slope',
    '项目阶段': 'project_phase',
    '标杆风机标志(0或1)': 'benchmark_flag',
    '遥测辅助状态判断启用': 'telemetry_aux_status_enable',
    'scada单风机UR': 'scada_ur',
    'scada风机图标': 'scada_icon',
    'scada数据冻结时间阈值': 'scada_freeze_threshold',
    'scada应用配置导入模板': 'scada_template',
    '预留字段': 'reserved_field',
    '风机图片文件名': 'turbine_image',
    '时区': 'timezone',
    '描述': 'description',
    '额定容量': 'rated_capacity',
    '机型': 'model',
    '外部厂家风机ID': 'external_turbine_id',
    '风机制造商': 'manufacturer',
    '切入风速': 'cut_in_wind_speed',
    '切出风速': 'cut_out_wind_speed',
    '修改时间': 'update_time',
    '关联进线间隔id': 'incoming_line_id',
    '关联出线间隔id': 'outgoing_line_id',
    '关联主变压器id': 'main_transformer_id',
    '投产状态': 'operation_status',
    '质保状态': 'warranty_status',
    '有功功率积分算电量': 'active_power_integral',
    '开启故障检修判断': 'fault_maintenance_enable',
    '实施阶段': 'implementation_phase',
    '次级分组': 'secondary_group',
    '风机编号': 'turbine_number',
    '风机阶段': 'turbine_phase',
    '首次并网时间': 'first_grid_time',
    '数据湖源': 'data_lake_source',
    'wsfd_local': 'wsfd_local',
    '叶轮直径': 'rotor_diameter'
}

# SQL语句定义
CREATE_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS info_turbine (
    id INT AUTO_INCREMENT PRIMARY KEY,
    turbine_id VARCHAR(50),
    wind_farm_id VARCHAR(50),
    turbine_name VARCHAR(200),
    turbine_name_en VARCHAR(200),
    uscada_alias VARCHAR(100),
    altitude DECIMAL(10, 4),
    latitude DECIMAL(12, 8),
    longitude DECIMAL(12, 8),
    main_control_version VARCHAR(100),
    feeder_id VARCHAR(50),
    phase_id VARCHAR(50),
    hub_height DECIMAL(10, 2),
    stat_start_date DATE,
    operation_date DATE,
    contract_power_curve_std VARCHAR(100),
    contract_power_curve_local VARCHAR(100),
    power_multiplier DECIMAL(10, 2),
    power_jump_slope DECIMAL(10, 2),
    project_phase INT,
    benchmark_flag TINYINT,
    telemetry_aux_status_enable VARCHAR(10),
    scada_ur VARCHAR(100),
    scada_icon VARCHAR(100),
    scada_freeze_threshold VARCHAR(100),
    scada_template VARCHAR(100),
    reserved_field VARCHAR(100),
    turbine_image VARCHAR(200),
    timezone VARCHAR(50),
    description TEXT,
    rated_capacity INT,
    model VARCHAR(100),
    external_turbine_id VARCHAR(100),
    manufacturer VARCHAR(100),
    cut_in_wind_speed DECIMAL(5, 2),
    cut_out_wind_speed DECIMAL(5, 2),
    update_time DATETIME,
    incoming_line_id VARCHAR(50),
    outgoing_line_id VARCHAR(50),
    main_transformer_id VARCHAR(50),
    operation_status VARCHAR(50),
    warranty_status VARCHAR(50),
    active_power_integral TINYINT,
    fault_maintenance_enable TINYINT,
    implementation_phase INT,
    secondary_group VARCHAR(100),
    turbine_number VARCHAR(50),
    turbine_phase VARCHAR(50),
    first_grid_time DATE,
    data_lake_source VARCHAR(100),
    wsfd_local VARCHAR(100),
    rotor_diameter INT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    INDEX idx_turbine_id (turbine_id),
    INDEX idx_wind_farm_id (wind_farm_id),
    INDEX idx_manufacturer (manufacturer),
    INDEX idx_model (model)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
"""

COUNT_TABLE_SQL = "SELECT COUNT(*) FROM info_turbine"
DROP_TABLE_SQL = "DROP TABLE IF EXISTS info_turbine"


class DatabaseConfig:
    """数据库配置类"""
    def __init__(self, host='192.168.50.234', port=4000, user='root', 
                 password='123456', database='wind_data', charset='utf8mb4'):
        self.host = host
        self.port = port
        self.user = user
        self.password = password
        self.database = database
        self.charset = charset
    
    def get_pymysql_config(self) -> Dict[str, Any]:
        """获取pymysql连接配置"""
        return {
            'host': self.host,
            'port': self.port,
            'user': self.user,
            'password': self.password,
            'database': self.database,
            'charset': self.charset
        }
    
    def get_sqlalchemy_url(self) -> str:
        """获取SQLAlchemy连接URL"""
        return f"mysql+pymysql://{self.user}:{self.password}@{self.host}:{self.port}/{self.database}?charset={self.charset}"


class TurbineDataImporter:
    """风机数据导入器"""
    
    def __init__(self, db_config: DatabaseConfig):
        """
        初始化导入器
        
        Args:
            db_config: 数据库配置对象
        """
        self.db_config = db_config
        self.engine = None
        self.pymysql_conn = None
        
    def connect_pymysql(self) -> pymysql.connections.Connection:
        """创建pymysql连接"""
        try:
            self.pymysql_conn = pymysql.connect(**self.db_config.get_pymysql_config())
            logger.info("pymysql连接成功")
            return self.pymysql_conn
        except Exception as e:
            logger.error(f"pymysql连接失败: {e}")
            raise
    
    def connect_sqlalchemy(self) -> create_engine:
        """创建SQLAlchemy引擎"""
        try:
            self.engine = create_engine(self.db_config.get_sqlalchemy_url())
            logger.info("SQLAlchemy引擎创建成功")
            return self.engine
        except Exception as e:
            logger.error(f"SQLAlchemy引擎创建失败: {e}")
            raise
    
    def close_connections(self):
        """关闭所有数据库连接"""
        if self.pymysql_conn:
            self.pymysql_conn.close()
            logger.info("pymysql连接已关闭")
        
        if self.engine:
            self.engine.dispose()
            logger.info("SQLAlchemy引擎已关闭")
    
    def create_table(self) -> bool:
        """
        创建风机信息表
        
        Returns:
            bool: 是否成功创建表
        """
        try:
            if not self.pymysql_conn:
                self.connect_pymysql()
            
            cursor = self.pymysql_conn.cursor()
            cursor.execute(CREATE_TABLE_SQL)
            self.pymysql_conn.commit()
            
            logger.info("表 info_turbine 创建成功或已存在")
            return True
            
        except Exception as e:
            logger.error(f"创建表时发生错误: {e}")
            self.pymysql_conn.rollback()
            return False
        finally:
            if 'cursor' in locals():
                cursor.close()
    
    def drop_table(self) -> bool:
        """
        删除风机信息表（用于重新创建）
        
        Returns:
            bool: 是否成功删除表
        """
        try:
            if not self.pymysql_conn:
                self.connect_pymysql()
            
            cursor = self.pymysql_conn.cursor()
            cursor.execute(DROP_TABLE_SQL)
            self.pymysql_conn.commit()
            
            logger.info("表 info_turbine 删除成功")
            return True
            
        except Exception as e:
            logger.error(f"删除表时发生错误: {e}")
            self.pymysql_conn.rollback()
            return False
        finally:
            if 'cursor' in locals():
                cursor.close()
    
    def load_csv_data(self, file_path: str) -> Optional[pd.DataFrame]:
        """
        加载CSV文件数据
        
        Args:
            file_path: CSV文件路径
            
        Returns:
            Optional[pd.DataFrame]: 加载的数据框，失败时返回None
        """
        try:
            logger.info(f"正在读取文件: {file_path}")
            
            # 读取CSV文件，指定编码
            df = pd.read_csv(file_path, encoding='utf-8')
            logger.info(f"成功读取CSV文件，共 {len(df)} 条记录")
            
            # 重命名列（中文表头转英文）
            df = df.rename(columns=HEADER_MAPPING)
            logger.info("表头已转换为英文")
            
            # 处理空值和异常值
            df = self._clean_data(df)
            
            # 数据类型转换
            df = self._convert_data_types(df)
            
            return df
            
        except FileNotFoundError:
            logger.error(f"错误：找不到文件 {file_path}")
            return None
        except Exception as e:
            logger.error(f"读取CSV文件时发生错误: {e}")
            return None
    
    def _clean_data(self, df: pd.DataFrame) -> pd.DataFrame:
        """清洗数据"""
        # 处理空字符串，转换为None
        df = df.replace(['', ' ', 'null', 'NULL'], None)
        
        # 删除全为空的列
        df = df.dropna(axis=1, how='all')
        
        return df
    
    def _convert_data_types(self, df: pd.DataFrame) -> pd.DataFrame:
        """转换数据类型"""
        
        # 处理日期和时间字段
        date_columns = ['stat_start_date', 'operation_date', 'first_grid_time']
        for col in date_columns:
            if col in df.columns:
                df[col] = pd.to_datetime(df[col], errors='coerce').dt.date
        
        # 处理update_time字段
        if 'update_time' in df.columns:
            df['update_time'] = pd.to_datetime(df['update_time'], errors='coerce')
        
        # 处理数值字段
        numeric_columns = [
            'altitude', 'latitude', 'longitude', 'hub_height', 
            'rated_capacity', 'cut_in_wind_speed', 'cut_out_wind_speed',
            'rotor_diameter', 'power_multiplier', 'power_jump_slope'
        ]
        
        for col in numeric_columns:
            if col in df.columns:
                df[col] = pd.to_numeric(df[col], errors='coerce')
        
        # 处理整数字段
        int_columns = ['project_phase', 'benchmark_flag', 'implementation_phase']
        for col in int_columns:
            if col in df.columns:
                df[col] = pd.to_numeric(df[col], errors='coerce').fillna(0).astype('Int64')
        
        # 处理布尔/标志字段
        flag_columns = ['active_power_integral', 'fault_maintenance_enable']
        for col in flag_columns:
            if col in df.columns:
                df[col] = pd.to_numeric(df[col], errors='coerce').fillna(0).astype('Int64')
        
        return df
    
    def import_to_mysql(self, df: pd.DataFrame, if_exists: str = 'append') -> bool:
        """
        导入数据到MySQL数据库
        
        Args:
            df: 数据框
            if_exists: 表存在时的处理方式 ('fail', 'replace', 'append')
            
        Returns:
            bool: 是否成功导入
        """
        try:
            if not self.engine:
                self.connect_sqlalchemy()
            
            logger.info("正在导入数据到MySQL数据库...")
            
            # 使用to_sql方法导入数据
            df.to_sql(
                name='info_turbine',
                con=self.engine,
                if_exists=if_exists,
                index=False,
                chunksize=1000,
                method='multi'  # 使用多值插入提高性能
            )
            
            logger.info(f"成功导入 {len(df)} 条记录到数据库表 info_turbine")
            return True
            
        except Exception as e:
            logger.error(f"导入数据到MySQL时发生错误: {e}")
            return False
    
    def get_record_count(self) -> Optional[int]:
        """
        获取表中的记录数
        
        Returns:
            Optional[int]: 记录数，失败时返回None
        """
        try:
            if not self.pymysql_conn:
                self.connect_pymysql()
            
            cursor = self.pymysql_conn.cursor()
            cursor.execute(COUNT_TABLE_SQL)
            count = cursor.fetchone()[0]
            cursor.close()
            
            logger.info(f"数据库表 info_turbine 中现有 {count} 条记录")
            return count
            
        except Exception as e:
            logger.error(f"获取记录数时发生错误: {e}")
            return None
    
    def get_table_info(self) -> Optional[Dict[str, Any]]:
        """
        获取表信息
        
        Returns:
            Optional[Dict[str, Any]]: 表信息字典
        """
        try:
            if not self.pymysql_conn:
                self.connect_pymysql()
            
            cursor = self.pymysql_conn.cursor()
            
            # 获取表结构
            cursor.execute("DESCRIBE info_turbine")
            columns = cursor.fetchall()
            
            # 获取索引信息
            cursor.execute("SHOW INDEX FROM info_turbine")
            indexes = cursor.fetchall()
            
            cursor.close()
            
            return {
                'columns': columns,
                'indexes': indexes
            }
            
        except Exception as e:
            logger.error(f"获取表信息时发生错误: {e}")
            return None
    
    def run_import_pipeline(self, csv_file_path: str, drop_if_exists: bool = False) -> bool:
        """
        运行完整的数据导入流程
        
        Args:
            csv_file_path: CSV文件路径
            drop_if_exists: 是否删除已存在的表
            
        Returns:
            bool: 整个流程是否成功
        """
        try:
            logger.info("开始执行风机数据导入程序...")
            
            # 步骤1: 创建数据库连接
            self.connect_pymysql()
            self.connect_sqlalchemy()
            
            # 步骤2: 处理表（如果需要删除则先删除）
            if drop_if_exists:
                if not self.drop_table():
                    return False
            
            # 步骤3: 创建表
            if not self.create_table():
                return False
            
            # 步骤4: 加载CSV数据
            df = self.load_csv_data(csv_file_path)
            if df is None:
                return False
            
            # 步骤5: 导入数据
            if not self.import_to_mysql(df, if_exists='append'):
                return False
            
            # 步骤6: 验证数据
            count = self.get_record_count()
            if count is not None:
                logger.info(f"数据导入验证成功，共导入 {len(df)} 条记录，数据库中现有 {count} 条记录")
            
            # 步骤7: 获取表信息（可选）
            table_info = self.get_table_info()
            if table_info:
                logger.info(f"表结构信息已获取，共有 {len(table_info['columns'])} 列")
            
            logger.info("风机数据导入程序执行完成！")
            return True
            
        except Exception as e:
            logger.error(f"数据导入流程执行失败: {e}")
            return False
        finally:
            # 清理连接
            self.close_connections()


def main():
    """主函数"""
    # 忽略警告
    warnings.filterwarnings('ignore')
    
    # 数据库配置
    db_config = DatabaseConfig(
        host='192.168.50.234',
        port=4000,
        user='root',
        password='123456',
        database='wind_data',
        charset='utf8mb4'
    )
    
    # CSV文件路径
    csv_file_path = './data/风机应用配置表_带叶轮直径.csv'
    
    # 创建导入器实例
    importer = TurbineDataImporter(db_config)
    
    # 运行导入流程
    # 如果表已经存在并且包含旧数据，可以设置 drop_if_exists=True 来重新创建表
    success = importer.run_import_pipeline(
        csv_file_path=csv_file_path,
        drop_if_exists=False  # 设置为True会删除已存在的表
    )
    
    if success:
        logger.info("数据导入成功完成！")
    else:
        logger.error("数据导入失败！")


if __name__ == "__main__":
    main()