通用Excel模板导入MySQL数据库的Python实现（附完整源码）

软件工程师文艺

1186人浏览 · 2025-06-24 14:25:20

软件工程师文艺 · 2025-06-24 14:25:20 发布

文章目录

一、技术实现原理
三、核心代码实现
四、扩展方向
五、总结

一、技术实现原理

数据类型自动映射机制
通过Pandas的dtypes属性获取Excel列的数据类型，建立从Python类型到MySQL类型的映射关系：

type_mapping = {
    'int64': 'INT',
    'float64': 'DECIMAL(10,2)',
    'datetime64': 'DATETIME',
    'object': 'VARCHAR(255)'
}

当检测到新的数据类型时，系统会自动扩展映射规则。例如发现布尔类型数据时，可追加映射’bool’: ‘TINYINT(1)’。

空值处理策略
采用差异化处理机制：

数值类型：填充0（整型）或0.0（浮点型）；
日期类型：填充pandas.NaT（时间戳）；
字符串类型：保留空字符串；
特殊类型：如布尔值填充False；

def _handle_missing_values(self, df):
    for col in df.columns:
        if df[col].dtype == 'object':
            df[col] = df[col].fillna('')
        elif df[col].dtype == 'int64':
            df[col] = df[col].fillna(0)
        elif df[col].dtype == 'float64':
            df[col] = df[col].fillna(0.0)
        elif 'datetime' in str(df[col].dtype):
            df[col] = df[col].fillna(pd.NaT)
    return df

自动建表机制
通过SQL模板动态生成建表语句，支持IF NOT EXISTS条件判断：

create_sql = f"CREATE TABLE IF NOT EXISTS `{table_name}` ({', '.join(columns)}) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;"

字段定义包含完整的数据类型声明和约束条件，支持自定义存储引擎和字符集。

二、程序执行时序

整体执行流程

二、程序执行时序
2. 初始化配置加载
   └── 读取YAML配置文件
   └── 建立数据库连接
   └── 配置日志系统

3. 数据导入过程
   └── 读取Excel文件
   └── 检测列数据类型
   └── 处理空值
   └── 创建目标表
   └── 插入数据记录

核心模块交互时序

三、核心代码实现

配置管理模块

# config.yaml
database:
  host: "localhost"
  port: 3306
  user: "root"
  password: "your_password"
  database: "test_db"
  table: "imported_data"
logging:
  level: "DEBUG"
  file: "excel_importer.log"

数据类型映射模块

def _detect_column_types(self, df):
    type_mapping = {
        'int64': 'INT',
        'float64': 'DECIMAL(10,2)',
        'datetime64': 'DATETIME',
        'object': 'VARCHAR(255)'
    }
    
    columns = []
    for col in df.columns:
        dtype = str(df[col].dtype)
        mysql_type = type_mapping.get(dtype, 'TEXT')
        columns.append(f"`{col}` {mysql_type}")
    return columns

数据处理模块

def import_data(self, excel_path, sheet_name=0):
    self.logger.info(f"开始导入文件: {excel_path}")
    
    # 读取Excel
    df = pd.read_excel(excel_path, sheet_name=sheet_name)
    df = self._handle_missing_values(df)
    
    # 获取表名和字段定义
    table_name = self.config['database']['table']
    columns = self._detect_column_types(df)
    
    # 自动创建表
    self.create_table(table_name, columns)
    
    # 插入数据
    insert_sql = f"INSERT INTO `{table_name}` ({','.join(f'`{col}`' for col in df.columns)}) VALUES ({','.join(['%s']*len(df.columns))})"
    
    try:
        for _, row in df.iterrows():
            self.cursor.execute(insert_sql, tuple(row))
        self.conn.commit()
        self.logger.info(f"导入完成，共 {len(df)} 条记录")
    except Exception as e:
        self.logger.error(f"数据插入失败: {str(e)}")
        self.conn.rollback()

四、扩展方向

性能优化
批量插入：使用executemany替代逐条插入

self.cursor.executemany(insert_sql, df.values.tolist())

扩展功能

支持多工作表导入：

def import_multiple_sheets(self, excel_path):
    sheet_names = pd.ExcelFile(excel_path).sheet_names
    for sheet in sheet_names:
        self.import_data(excel_path, sheet)

异常处理

增加重试机制：

def _execute_with_retry(self, sql, params=None, retry=3):
    for i in range(retry):
        try:
            if params:
                self.cursor.execute(sql, params)
            else:
                self.cursor.execute(sql)
            return True
        except Exception as e:
            self.logger.warning(f"操作失败，第{i+1}次重试: {str(e)}")
            time.sleep(2)
    return False