Python中如何实现将Excel数据导入MySQL的小工具

我来写一个完整的Python脚本，可以直接将Excel数据导入MySQL数据库。

import pandas as pd
import mysql.connector
from mysql.connector import Error
import argparse
import sys

class ExcelToMySQL:
    def __init__(self, host, user, password, database):
        """初始化数据库连接参数"""
        self.host = host
        self.user = user
        self.password = password
        self.database = database
        self.connection = None
        
    def connect_to_mysql(self):
        """建立MySQL数据库连接"""
        try:
            self.connection = mysql.connector.connect(
                host=self.host,
                user=self.user,
                password=self.password,
                database=self.database
            )
            if self.connection.is_connected():
                print(f"成功连接到MySQL数据库: {self.database}")
                return True
        except Error as e:
            print(f"连接数据库失败: {e}")
            return False
    
    def read_excel_file(self, excel_path, sheet_name=0):
        """读取Excel文件"""
        try:
            # 读取Excel文件
            df = pd.read_excel(excel_path, sheet_name=sheet_name)
            print(f"成功读取Excel文件: {excel_path}")
            print(f"数据形状: {df.shape}")
            print(f"列名: {list(df.columns)}")
            return df
        except Exception as e:
            print(f"读取Excel文件失败: {e}")
            return None
    
    def create_table_from_dataframe(self, table_name, df):
        """根据DataFrame创建MySQL表"""
        cursor = self.connection.cursor()
        
        # 生成创建表的SQL语句
        columns = []
        for col in df.columns:
            # 简单推断数据类型
            dtype = str(df[col].dtype)
            if 'int' in dtype:
                sql_type = 'INT'
            elif 'float' in dtype:
                sql_type = 'FLOAT'
            elif 'datetime' in dtype:
                sql_type = 'DATETIME'
            else:
                # 获取最大字符串长度
                max_len = df[col].astype(str).str.len().max()
                sql_type = f'VARCHAR({int(max_len) + 10})'
            columns.append(f"`{col}` {sql_type}")
        
        create_table_sql = f"""
        CREATE TABLE IF NOT EXISTS `{table_name}` (
            id INT AUTO_INCREMENT PRIMARY KEY,
            {', '.join(columns)}
        )
        """
        
        try:
            cursor.execute(create_table_sql)
            print(f"表 '{table_name}' 创建成功或已存在")
            return True
        except Error as e:
            print(f"创建表失败: {e}")
            return False
        finally:
            cursor.close()
    
    def insert_data_to_table(self, table_name, df):
        """将DataFrame数据插入到MySQL表"""
        cursor = self.connection.cursor()
        
        # 准备插入数据的SQL语句
        columns = ', '.join([f"`{col}`" for col in df.columns])
        placeholders = ', '.join(['%s'] * len(df.columns))
        insert_sql = f"INSERT INTO `{table_name}` ({columns}) VALUES ({placeholders})"
        
        try:
            # 将DataFrame转换为元组列表
            data_tuples = [tuple(x) for x in df.to_numpy()]
            
            # 批量插入数据
            cursor.executemany(insert_sql, data_tuples)
            self.connection.commit()
            
            print(f"成功插入 {cursor.rowcount} 条数据到表 '{table_name}'")
            return True
        except Error as e:
            print(f"插入数据失败: {e}")
            self.connection.rollback()
            return False
        finally:
            cursor.close()
    
    def import_excel_to_mysql(self, excel_path, table_name, sheet_name=0):
        """主函数：导入Excel数据到MySQL"""
        # 连接数据库
        if not self.connect_to_mysql():
            return False
        
        # 读取Excel文件
        df = self.read_excel_file(excel_path, sheet_name)
        if df is None:
            return False
        
        # 创建表
        if not self.create_table_from_dataframe(table_name, df):
            return False
        
        # 插入数据
        if not self.insert_data_to_table(table_name, df):
            return False
        
        return True
    
    def close_connection(self):
        """关闭数据库连接"""
        if self.connection and self.connection.is_connected():
            self.connection.close()
            print("数据库连接已关闭")

def main():
    # 设置命令行参数
    parser = argparse.ArgumentParser(description='将Excel数据导入MySQL数据库')
    parser.add_argument('--host', default='localhost', help='MySQL主机地址')
    parser.add_argument('--user', required=True, help='MySQL用户名')
    parser.add_argument('--password', required=True, help='MySQL密码')
    parser.add_argument('--database', required=True, help='数据库名')
    parser.add_argument('--excel', required=True, help='Excel文件路径')
    parser.add_argument('--table', required=True, help='目标表名')
    parser.add_argument('--sheet', default=0, help='Excel工作表名或索引')
    
    args = parser.parse_args()
    
    # 创建导入器实例
    importer = ExcelToMySQL(
        host=args.host,
        user=args.user,
        password=args.password,
        database=args.database
    )
    
    try:
        # 执行导入
        success = importer.import_excel_to_mysql(
            excel_path=args.excel,
            table_name=args.table,
            sheet_name=args.sheet
        )
        
        if success:
            print("导入完成！")
        else:
            print("导入失败！")
            sys.exit(1)
            
    finally:
        # 关闭连接
        importer.close_connection()

if __name__ == "__main__":
    main()

使用方法：

1. 安装依赖：

pip install pandas mysql-connector-python openpyxl

1. 命令行运行：

# 基本用法
python excel_to_mysql.py --user root --password 123456 --database testdb --excel data.xlsx --table my_table

# 完整参数
python excel_to_mysql.py --host localhost --user root --password 123456 --database testdb --excel data.xlsx --table my_table --sheet "Sheet1"

功能特点：

• 自动读取Excel文件（支持.xlsx和.xls）
• 根据Excel数据结构自动创建MySQL表
• 批量插入数据，提高导入效率
• 支持命令行参数，方便集成到自动化流程
• 包含错误处理和事务回滚

使用建议： 先测试小文件，确认数据结构正确后再导入大量数据。