AcidMap/scripts/import_parameters.py

"""
参数数据导入脚本
@description: 从Excel文件读取Parameters参数数据并导入到Parameters表
"""

import os
import sys
import pandas as pd
import logging
from datetime import datetime

# 添加项目根目录到Python路径
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from app.database import engine, SessionLocal
from app.models.parameters import Parameters  # 确保你已创建Parameters模型

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

class ParametersDataImporter:
    """
    参数数据导入器

    @description: 从Excel文件读取参数数据并导入到Parameters表
    """

    def __init__(self, excel_path, sheet_name='Parameters'):
        """
        初始化导入器

        @param {str} excel_path - Excel文件路径
        @param {str} sheet_name - Sheet名称，默认为'Parameters'
        """
        self.excel_path = excel_path
        self.sheet_name = sheet_name

        # 定义参数列的默认值映射
        self.default_values = {
            'F1': 0.6,
            'F2': 0.85,
            'F3': 0.05,
            'F4': 0.158,
            'F5': 0.06,
            'F6': 0.065,
            'F7': 0.6,
            'F8': 0.25,
            'F9': 0.35,
            'F10': 0.25,
            'F11': 800,
            'NF': 0.05,
            'PF': 0.158,
            'KF': 0.06,
            'CF': 0.065,
            'OF': 0.6,
            'P': 0.25,
            'FF': 0.35,
            'AF': 0.25,
            'Area': '韶关'
        }

        # 定义必需的最小参数列
        self.required_columns = ['Area']

    def read_excel_data(self):
        """
        读取Excel文件数据

        @returns: DataFrame 读取的数据
        """
        try:
            logger.info(f"开始读取Excel文件: {self.excel_path}")
            logger.info(f"Sheet名称: {self.sheet_name}")

            # 检查文件是否存在
            if not os.path.exists(self.excel_path):
                raise FileNotFoundError(f"Excel文件不存在: {self.excel_path}")

            # 读取Excel文件
            df = pd.read_excel(self.excel_path, sheet_name=self.sheet_name)

            logger.info(f"成功读取数据，共 {len(df)} 行")
            logger.info(f"数据列: {list(df.columns)}")

            # 显示前几行数据供确认
            logger.info("前5行数据预览:")
            logger.info(df.head().to_string())

            return df

        except Exception as e:
            logger.error(f"读取Excel文件失败: {str(e)}")
            raise

    def validate_data(self, df):
        """
        验证数据格式和完整性

        @param {DataFrame} df - 要验证的数据
        @returns: DataFrame 验证后的数据
        """
        try:
            logger.info("开始验证数据...")

            # 检查必需的列是否存在
            missing_columns = [col for col in self.required_columns if col not in df.columns]

            if missing_columns:
                raise ValueError(f"缺少必需的列: {missing_columns}")

            # 确保数据文件包含所有参数列
            for column in self.default_values.keys():
                if column not in df.columns:
                    logger.warning(f"列 {column} 不存在于Excel文件中，将使用默认值 {self.default_values[column]}")
                    df[column] = self.default_values[column]

            # 验证数值列的数据类型
            numeric_columns = list(self.default_values.keys())
            numeric_columns.remove('Area')  # Area不是数值类型

            for col in numeric_columns:
                # 尝试转换为数值类型
                df[col] = pd.to_numeric(df[col], errors='coerce')

                # 检查是否有无效值
                if df[col].isnull().any():
                    invalid_rows = df[df[col].isnull()]
                    logger.warning(f"列 {col} 中有无效值，行号: {list(invalid_rows.index)}")

                    # 填充默认值
                    logger.info(f"将为无效行填充默认值 {self.default_values[col]}")
                    df[col] = df[col].fillna(self.default_values[col])

            # 确保Area列是字符串类型
            if 'Area' in df.columns:
                df['Area'] = df['Area'].astype(str)

            logger.info(f"数据验证完成，有效数据 {len(df)} 行")

            return df

        except Exception as e:
            logger.error(f"数据验证失败: {str(e)}")
            raise

    def import_data(self, df):
        """
        将数据导入到数据库

        @param {DataFrame} df - 要导入的数据
        """
        try:
            logger.info("开始导入数据到数据库...")

            # 创建数据库会话
            db = SessionLocal()

            try:
                # 检查是否有重复数据
                existing_count = db.query(Parameters).count()
                logger.info(f"数据库中现有参数记录: {existing_count} 条")

                # 如果已有数据，先删除现有数据
                if existing_count > 0:
                    db.query(Parameters).delete()
                    logger.info(f"已删除 {existing_count} 条现有参数记录")

                # 获取总行数
                total_rows = len(df)
                imported_count = 0
                skipped_count = 0

                # 使用ORM模型创建对象并插入
                for i, row in df.iterrows():
                    try:
                        # 创建参数对象
                        param = Parameters(
                            f1=row['F1'],
                            f2=row['F2'],
                            f3=row['F3'],
                            f4=row['F4'],
                            f5=row['F5'],
                            f6=row['F6'],
                            f7=row['F7'],
                            f8=row['F8'],
                            f9=row['F9'],
                            f10=row['F10'],
                            f11=row['F11'],
                            nf=row['NF'],
                            pf=row['PF'],
                            kf=row['KF'],
                            cf=row['CF'],
                            of=row['OF'],
                            p=row['P'],
                            ff=row['FF'],
                            af=row['AF'],
                            area=row['Area']
                        )

                        # 添加到会话
                        db.add(param)
                        imported_count += 1

                        # 每100条提交一次
                        if imported_count % 100 == 0:
                            db.commit()
                            logger.info(f"已导入 {imported_count}/{total_rows} 条数据")

                    except Exception as e:
                        logger.warning(f"跳过行 {i}: {str(e)}")
                        skipped_count += 1
                        db.rollback()  # 回滚当前会话

                # 提交剩余的数据
                db.commit()

                logger.info(f"参数导入完成! 成功导入 {imported_count} 条数据, 跳过 {skipped_count} 条无效数据")

                # 验证导入结果
                final_count = db.query(Parameters).count()
                logger.info(f"导入后数据库参数记录: {final_count} 条")

            except Exception as e:
                db.rollback()
                logger.error(f"数据导入失败，已回滚: {str(e)}")
                raise
            finally:
                db.close()

        except Exception as e:
            logger.error(f"数据导入过程失败: {str(e)}")
            raise

    def run_import(self):
        """
        执行完整的导入流程
        """
        try:
            logger.info("=" * 60)
            logger.info("开始参数数据导入流程")
            logger.info("=" * 60)

            # 1. 读取Excel数据
            df = self.read_excel_data()

            # 2. 验证数据
            df = self.validate_data(df)

            # 3. 导入数据
            self.import_data(df)

            logger.info("=" * 60)
            logger.info("参数数据导入流程完成!")
            logger.info("=" * 60)

        except Exception as e:
            logger.error(f"导入流程失败: {str(e)}")
            raise

def main():
    """
    主函数
    """
    # Excel文件路径
    excel_path = r"D:\destkop\数据库对应数据.xlsx"  # 或你的文件路径
    sheet_name = "Parameters"  # 确保你的Excel文件中有这个sheet

    try:
        # 创建导入器并执行导入
        importer = ParametersDataImporter(excel_path, sheet_name)
        importer.run_import()

    except Exception as e:
        logger.error(f"程序执行失败: {str(e)}")
        sys.exit(1)

if __name__ == "__main__":
    main()