一键批量导入Excel数据到MySQL的桌面工具，超级好用！

最近有不少朋友问我，有没有什么办法能快速把Excel文件里的数据批量导入到MySQL数据库？毕竟手动复制粘贴太慢了，于是我开发了一款小巧易用的桌面工具，简单分享给大家。

这个工具可以直接一次性导入整个文件夹里所有的.xlsx表格文件，非常方便。

工具的具体使用方法如下：

① 修改配置文件

解压工具后，会看到一个名叫 配置.ini 的文件，打开这个配置文件，根据你的实际需求修改一下里面的信息，主要包括：

• 数据库连接参数（主机、端口、用户名、密码、数据库名）

• 数据库对应表名称

• Excel文件的列名与数据库字段的对应关系

修改保存即可完成设置，非常简单。

② 数据库连接测试

运行工具后，直接点击工具界面上的『测试连接』按钮，确认数据库能正常连接后再进行下一步。

③ 导入Excel数据

点击『导入数据』按钮，选择你存放 .xlsx 文件的文件夹，工具会自动识别并导入该文件夹下所有符合条件的Excel文件，不需要逐个文件手动处理。

工具会实时展示导入进度、当前文件进度，以及整体导入状态，界面清晰易懂，非常直观。

工具特色功能：

• 支持通过ini配置文件，灵活设置Excel表列与数据库字段的映射关系。

• 提供详细的日志记录功能，方便问题排查。

• 自动批量导入，无需额外干预。

• 支持导入过程中实时显示进度与状态，方便用户直观看到导入情况。

工具源码技术栈：

• 界面开发：基于Python的tkinter实现。

• 数据处理：利用pandas进行Excel数据的读取与处理。

• 数据库连接：使用mysql.connector模块完成与MySQL数据库的交互。

• 配置管理：通过configparser实现ini文件的读取与参数配置。

• 日志管理：采用Python的logging库，实现了日志记录与错误处理。

源码内容：

import tkinter as tk
from tkinter import ttk, filedialog, messagebox
import threading
import pandas as pd
import mysql.connector
import os
import logging
from logging.handlers import RotatingFileHandler
import configparser

# 初始化日志设置
log = logging.getLogger("Excel2MySQL")
log.setLevel(logging.INFO)
log_handler = RotatingFileHandler('data_import.log', maxBytes=1024*1024, backupCount=3)
log_format = logging.Formatter('%(asctime)s [%(levelname)s] - %(message)s')
log_handler.setFormatter(log_format)
log.addHandler(log_handler)

class ExcelToDBTool:
def __init__(self, root_window):
self.root = root_window
self.root.title("Excel数据批量导入MySQL工具")

# 从配置文件加载数据库参数和字段映射关系
self.load_configuration()

# 搭建主界面
self.setup_interface()

def load_configuration(self):
cfg = configparser.ConfigParser()
cfg.read('配置.ini', encoding='utf-8')

# 数据库配置信息加载
self.db_info = {
'host': cfg.get('Database', 'host'),
'port': cfg.getint('Database', 'port'),
'user': cfg.get('Database', 'user'),
'password': cfg.get('Database', 'password'),
'database': cfg.get('Database', 'name'),
}

# 读取数据库表名
self.table = cfg.get('Table', 'name')

# Excel列和数据库字段映射关系
self.excel_cols = cfg.get('ExcelColumns', 'columns').split(',')
self.db_cols = cfg.get('DatabaseColumns', 'columns').split(',')

def setup_interface(self):
# 数据库连接测试区
db_frame = ttk.LabelFrame(self.root, text="MySQL连接配置")
db_frame.grid(row=0, column=0, padx=10, pady=10, sticky='ew')

ttk.Button(db_frame, text="测试数据库连接", command=self.verify_db_connection)\
.grid(row=0, column=0, pady=5, padx=5, sticky='w')
self.conn_status = ttk.Label(db_frame, text="等待连接测试...")
self.conn_status.grid(row=0, column=1, padx=5)

ttk.Label(db_frame, text=f"数据库: {self.db_info['database']}").grid(row=1, column=0, sticky='w', padx=5)
ttk.Label(db_frame, text=f"表名: {self.table}").grid(row=2, column=0, sticky='w', padx=5)

# 文件夹选择区
src_frame = ttk.LabelFrame(self.root, text="Excel数据源")
src_frame.grid(row=1, column=0, padx=10, pady=10, sticky='ew')

ttk.Button(src_frame, text="选择数据文件夹", command=self.choose_folder)\
.grid(row=0, column=0, pady=5, padx=5, sticky='w')
self.folder_label = ttk.Label(src_frame, text="尚未选择数据文件夹")
self.folder_label.grid(row=1, column=0, padx=5, sticky='w')

# Excel与数据库映射关系展示
mapping_frame = ttk.LabelFrame(self.root, text="数据字段映射关系")
mapping_frame.grid(row=2, column=0, padx=10, pady=10, sticky='ew')

self.mapping_tree = ttk.Treeview(mapping_frame, columns=("Excel列", "数据库字段"), show="headings")
self.mapping_tree.heading("Excel列", text="Excel列")
self.mapping_tree.heading("数据库字段", text="数据库字段")
self.mapping_tree.pack(fill='both', expand=True, padx=5, pady=5)

for e_col, d_col in zip(self.excel_cols, self.db_cols):
self.mapping_tree.insert("", "end", values=(e_col.strip(), d_col.strip()))

# 导入按钮与进度展示
ttk.Button(self.root, text="开始导入数据", command=self.initiate_import, state='disabled')\
.grid(row=3, column=0, pady=10, padx=10, sticky='w')

self.progress_label = ttk.Label(self.root, text="未开始导入")
self.progress_label.grid(row=4, column=0, sticky='w', padx=10)

self.progress_bar = ttk.Progressbar(self.root, length=250)
self.progress_bar.grid(row=5, column=0, pady=10, padx=10, sticky='w')

def verify_db_connection(self):
try:
conn = mysql.connector.connect(**self.db_info)
conn.close()
self.conn_status.config(text="连接成功！", foreground='green')
log.info("数据库连接测试成功！")
messagebox.showinfo("数据库连接", "连接成功！")
except Exception as err:
self.conn_status.config(text="连接失败！", foreground='red')
log.error(f"数据库连接失败: {err}")
messagebox.showerror("数据库连接", f"连接失败: {err}")

def choose_folder(self):
selected_dir = filedialog.askdirectory()
if selected_dir:
self.folder_label.config(text=selected_dir)
self.root.children['!button4'].config(state='normal')
self.selected_directory = selected_dir

def initiate_import(self):
threading.Thread(target=self.batch_import).start()

def batch_import(self):
excel_files = [f for f in os.listdir(self.selected_directory) if f.endswith('.xlsx')]
total_files = len(excel_files)
imported_rows = 0

try:
db_conn = mysql.connector.connect(**self.db_info)
cursor = db_conn.cursor()

sql_insert = f"INSERT INTO {self.table} ({', '.join(self.db_cols)}) VALUES ({', '.join(['%s'] * len(self.db_cols))})"

for idx, excel_file in enumerate(excel_files, start=1):
file_path = os.path.join(self.selected_directory, excel_file)
data_frame = pd.read_excel(file_path, usecols=self.excel_cols)

records = data_frame.where(pd.notna(data_frame), None).values.tolist()

cursor.executemany(sql_insert, records)
db_conn.commit()

imported_rows += len(records)
self.progress_label.config(text=f"已导入文件 {idx}/{total_files}，共计 {imported_rows} 行")
self.progress_bar['value'] = (idx / total_files) * 100
self.root.update_idletasks()

log.info(f"文件 {excel_file} 导入成功，共导入 {len(records)} 行数据")

messagebox.showinfo("导入完成", f"所有文件导入成功，共计 {imported_rows} 行数据！")
log.info("批量数据导入完成。")

except Exception as import_err:
log.error(f"导入过程中出现错误：{import_err}")
messagebox.showerror("导入失败", f"导入过程中出现错误: {import_err}")

finally:
cursor.close()
db_conn.close()

if __name__ == "__main__":
main_win = tk.Tk()
app = ExcelToDBTool(main_win)
main_win.mainloop()

下载地址：


隐藏内容，解锁需要先评论本文
评论后刷新解锁