Excel批量导入MySQL的桌面工具，配置灵活，带GUI界面

今天分享一个我自己常用的工具：Excel数据批量写入MySQL的小程序。它的核心用途就是——选定文件夹，一键把所有.xlsx文件里的数据，自动导进MySQL表中。

基本使用方式

1. 配置.ini 文件
   程序启动前先改下 配置.ini，填写数据库连接信息（主机、端口、用户名、密码），以及目标表名。
   还要设置下字段映射，Excel表头列 和 数据库字段 一一对应好。

📷 [Excel数据导入MySQL工具 使用图 1]

2. 测试连接
   软件界面上点一下“测试连接”，会返回是否成功接入数据库。

3. 开始导入数据
   点击“导入数据”按钮后，选择你存放 .xlsx 文件的文件夹，软件会自动扫描并处理所有文件。

📷 [Excel数据导入MySQL工具 使用图 2]

程序实现说明（源码结构）

这个工具用的是 Python + tkinter 做的图形界面，核心依赖包括：

• pandas：处理Excel表格；

• mysql.connector：用于数据库写入；

• configparser：加载 .ini 配置；

• logging：完整日志记录机制；

• threading：导入数据过程异步执行，不阻塞UI。

程序结构清晰，包含以下几个关键模块：

• 数据库连接与测试模块

• Excel列名与数据库字段比对逻辑

• 导入进度条显示（每个文件、总进度）

• 数据清洗与写入、异常捕捉与事务控制

• 失败时自动重试机制

功能细节亮点

• 每个 .xlsx 文件读取前会先验证列名是否和配置中一致，防止错导。

• 批量写入采用事务控制，失败会自动回滚，并支持重试和单条补救。

• 所有关键操作都写入日志文件，方便查错。

• GUI操作直观，适合没有写代码经验的人也能快速用上。

# Excel批量导入MySQL工具（改写版本）
# 作者：韩仔（重构注释+UI+变量命名+功能文字，保留逻辑结构）

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

# 初始化日志记录器
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
log_handler = RotatingFileHandler('app.log', maxBytes=1000000, backupCount=5)
log_formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
log_handler.setFormatter(log_formatter)
logger.addHandler(log_handler)

class ExcelImporterApp:
def __init__(self, root):
self.root = root
root.title("Excel导入MySQL工具 - By 小吴")

# 加载配置
cfg = configparser.ConfigParser()
cfg.read('配置.ini', encoding='utf-8')

self.db_config = {
'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')
self.excel_cols = cfg.get('ExcelColumns', 'columns').split(',')
self.db_cols = cfg.get('DatabaseColumns', 'columns').split(',')

self.build_ui()

def build_ui(self):
# 数据库连接配置区域
db_frame = ttk.LabelFrame(self.root, text="数据库设置")
db_frame.grid(row=0, column=0, sticky="ew", padx=5, pady=5)

ttk.Button(db_frame, text="测试连接", command=self.test_db).grid(row=0, column=0, pady=5, sticky="w")
self.conn_status = ttk.Label(db_frame, text="未连接")
self.conn_status.grid(row=0, column=1, sticky="w")

ttk.Label(db_frame, text="数据库:").grid(row=1, column=0, sticky="w")
ttk.Label(db_frame, text=self.db_config['database']).grid(row=1, column=1, sticky="w")

ttk.Label(db_frame, text="目标表:").grid(row=2, column=0, sticky="w")
ttk.Label(db_frame, text=self.table).grid(row=2, column=1, sticky="w")

# 数据文件选择区
source_frame = ttk.LabelFrame(self.root, text="Excel文件路径")
source_frame.grid(row=1, column=0, sticky="ew", padx=5, pady=5)

ttk.Button(source_frame, text="选择文件夹", command=self.select_dir).grid(row=0, column=0, pady=5)
self.dir_label = ttk.Label(source_frame, text="尚未选择")
self.dir_label.grid(row=1, column=0, sticky="w")

# 映射关系展示
map_frame = ttk.LabelFrame(self.root, text="列映射预览")
map_frame.grid(row=2, column=0, sticky="ew", padx=5, pady=5)

self.map_tree = ttk.Treeview(map_frame, columns=("Excel列", "数据库字段"), show="headings")
self.map_tree.heading("Excel列", text="Excel列")
self.map_tree.heading("数据库字段", text="数据库字段")
for ec, dc in zip(self.excel_cols, self.db_cols):
self.map_tree.insert('', 'end', values=(ec, dc))
self.map_tree.pack(fill="both", expand=True)

# 控制按钮
self.import_btn = ttk.Button(self.root, text="开始导入", command=self.begin_import, state="disabled")
self.import_btn.grid(row=3, column=0, sticky="w", pady=5)
self.import_status = ttk.Label(self.root, text="等待中")
self.import_status.grid(row=4, column=0, sticky="w")

# 进度指示
self.file_bar = ttk.Progressbar(self.root, orient="horizontal", length=200, mode="determinate")
self.file_bar.grid(row=5, column=0, sticky="w")
self.file_status = ttk.Label(self.root, text="文件进度: ")
self.file_status.grid(row=6, column=0, sticky="w")

self.total_bar = ttk.Progressbar(self.root, orient="horizontal", length=200, mode="determinate")
self.total_bar.grid(row=7, column=0, sticky="w")
self.total_status = ttk.Label(self.root, text="总进度: ")
self.total_status.grid(row=8, column=0, sticky="w")

def test_db(self):
try:
conn = mysql.connector.connect(**self.db_config)
conn.close()
self.conn_status.config(text="已连接")
logger.info("数据库连接成功")
messagebox.showinfo("成功", "成功连接到数据库")
except Exception as e:
self.conn_status.config(text="连接失败")
logger.error(f"数据库连接异常: {e}")
messagebox.showerror("连接失败", str(e))

def select_dir(self):
folder = filedialog.askdirectory()
if folder:
self.dir_label.config(text=folder)
self.verify_excel_columns(folder)

def verify_excel_columns(self, folder):
first_file = next((f for f in os.listdir(folder) if f.endswith('.xlsx')), None)
if not first_file:
return
try:
df = pd.read_excel(os.path.join(folder, first_file), usecols=self.excel_cols, nrows=1)
if set(self.excel_cols).issubset(df.columns):
self.import_btn.config(state="normal")
self.import_status.config(text="可以导入")
else:
messagebox.showwarning("列名不匹配", "Excel列名与配置不一致")
except Exception as e:
messagebox.showerror("读取失败", str(e))

def begin_import(self):
self.import_btn.config(state="disabled")
self.import_status.config(text="处理中...")
folder = self.dir_label.cget("text")
self.files = [f for f in os.listdir(folder) if f.endswith('.xlsx')]
self.total_rows = sum(len(pd.read_excel(os.path.join(folder, f), usecols=self.excel_cols)) for f in self.files)
self.done_rows = 0

self.db = mysql.connector.connect(**self.db_config)
self.insert_sql = f"INSERT INTO {self.table} ({', '.join(self.db_cols)}) VALUES ({', '.join(['%s'] * len(self.db_cols))})"

threading.Thread(target=self.run_import_thread, args=(folder,)).start()

def run_import_thread(self, folder):
cursor = self.db.cursor()
try:
for idx, file in enumerate(self.files):
path = os.path.join(folder, file)
df = pd.read_excel(path, usecols=self.excel_cols)
data = df.where(pd.notna(df), None).values.tolist()
total = len(data)
inserted = 0

for i in range(0, total, 1000):
batch = data[i:i+1000]
try:
cursor.execute("START TRANSACTION")
cursor.executemany(self.insert_sql, batch)
cursor.execute("COMMIT")
inserted += len(batch)
except Exception as e:
logger.warning(f"批次插入失败：{e}，尝试单行补救")
cursor.execute("ROLLBACK")
for row in batch:
try:
cursor.execute("START TRANSACTION")
cursor.execute(self.insert_sql, row)
cursor.execute("COMMIT")
inserted += 1
except:
cursor.execute("ROLLBACK")

self.file_status.config(text=f"文件: {file} - {inserted}/{total}")
self.file_bar['value'] = (inserted * 100) / total
self.done_rows += inserted
self.total_status.config(text=f"总进度: {self.done_rows}/{self.total_rows}")
self.total_bar['value'] = (self.done_rows * 100) / self.total_rows
self.root.update_idletasks()

messagebox.showinfo("完成", "所有数据已导入！")
finally:
cursor.close()
self.db.close()

if __name__ == '__main__':
root = tk.Tk()
app = ExcelImporterApp(root)
root.mainloop()

附件下载：

📦 工具已打包为可执行文件，直接双击运行：
Excel数据导入数据库MySQL工具.exe

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