Python项目源码57：数据格式转换工具1.0（csv+json+excel+sqlite3）
# -*- coding: utf-8 -*-
# @Author : 小红牛
# 微信公众号：WdPython
import os
import tkinter as tk
from tkinter import ttk, filedialog, messagebox
from tkinter.simpledialog import askstring
import sqlite3
import pandas as pd
from pandastable import Table
class DataConverterApp:
def __init__(self, root):
self.root = root
self.root.title("智能数据转换工具1.0")
self.root.geometry("1200x700")
# 初始化变量
self.source_path = tk.StringVar()
self.target_path = tk.StringVar()
self.table_name = tk.StringVar()
self.conversion_type = tk.StringVar(value="excel2json")
# 文件类型映射
self.file_extensions = {
"excel": ("Excel文件", ".xlsx"),
"json": ("JSON文件", ".json"),
"csv": ("CSV文件", ".csv"),
"sqlite": ("SQLite数据库", ".db")
}
# 创建界面
self.create_widgets()
self.setup_preview_table()
def setup_preview_table(self):
"""初始化数据预览表格"""
self.preview_frame = ttk.Frame(self.preview_container)
self.preview_frame.pack(fill=tk.BOTH, expand=True)
self.ptable = Table(self.preview_frame, showtoolbar=False, showstatusbar=True)
self.ptable.show()
def create_widgets(self):
# 主容器
main_frame = ttk.Frame(self.root, padding=20)
main_frame.pack(fill=tk.BOTH, expand=True)
# 左侧控制面板
control_frame = ttk.Frame(main_frame, width=400)
control_frame.pack(side=tk.LEFT, fill=tk.Y)
# 右侧预览面板
self.preview_container = ttk.LabelFrame(main_frame, text="数据预览", padding=10)
self.preview_container.pack(side=tk.RIGHT, fill=tk.BOTH, expand=True)
# 转换类型选择
type_frame = ttk.LabelFrame(control_frame, text="转换类型", padding=10)
type_frame.pack(fill=tk.X, pady=5)
conversion_types = [
("Excel → JSON", "excel2json"),
("JSON → Excel", "json2excel"),
("CSV → JSON", "csv2json"),
("JSON → CSV", "json2csv"),
("SQLite → Excel", "sqlite2excel"),
("Excel → SQLite", "excel2sqlite"),
("SQLite → CSV", "sqlite2csv"),
("CSV → SQLite", "csv2sqlite")
]
for text, value in conversion_types:
rb = ttk.Radiobutton(type_frame, text=text, variable=self.conversion_type,
value=value, command=self.update_ui)
rb.pack(anchor=tk.W, pady=2)
# 源文件设置
source_frame = ttk.LabelFrame(control_frame, text="源文件设置", padding=10)
source_frame.pack(fill=tk.X, pady=5)
ttk.Label(source_frame, text="源文件路径:").pack(anchor=tk.W)
ttk.Entry(source_frame, textvariable=self.source_path, width=50).pack(side=tk.LEFT, fill=tk.X, expand=True)
ttk.Button(source_frame, text="浏览...", command=self.browse_source).pack(side=tk.RIGHT)
# 目标文件设置
target_frame = ttk.LabelFrame(control_frame, text="目标设置", padding=10)
target_frame.pack(fill=tk.X, pady=5)
ttk.Label(target_frame, text="目标路径:").pack(anchor=tk.W)
ttk.Entry(target_frame, textvariable=self.target_path, width=50).pack(side=tk.LEFT, fill=tk.X, expand=True)
ttk.Button(target_frame, text="浏览...", command=self.browse_target).pack(side=tk.RIGHT)
# 数据库表名设置
self.table_frame = ttk.LabelFrame(control_frame, text="数据库设置", padding=10)
ttk.Label(self.table_frame, text="表名:").pack(side=tk.LEFT)
ttk.Entry(self.table_frame, textvariable=self.table_name).pack(side=tk.LEFT, fill=tk.X, expand=True)
# 操作按钮
btn_frame = ttk.Frame(control_frame)
btn_frame.pack(fill=tk.X, pady=10)
ttk.Button(btn_frame, text="开始转换", command=self.start_conversion).pack(side=tk.LEFT, padx=5)
ttk.Button(btn_frame, text="清空预览", command=self.clear_preview).pack(side=tk.LEFT, padx=5)
ttk.Button(btn_frame, text="退出", command=self.root.quit).pack(side=tk.RIGHT, padx=5)
# 绑定路径变化事件
self.source_path.trace_add("write", self.update_preview)
self.table_name.trace_add("write", self.update_preview)
self.update_ui()
def get_conversion_info(self):
"""获取当前转换类型信息"""
ct = self.conversion_type.get()
source_type = ct.split("2")[0]
target_type = ct.split("2")[1]
return source_type, target_type
def update_ui(self):
"""更新界面元素"""
source_type, target_type = self.get_conversion_info()
# 更新数据库设置可见性
if "sqlite" in self.conversion_type.get():
self.table_frame.pack(fill=tk.X, pady=5)
else:
self.table_frame.pack_forget()
# 自动更新目标路径后缀
current_target = self.target_path.get()
if current_target:
base, _ = os.path.splitext(current_target)
new_ext = self.file_extensions[target_type][1]
self.target_path.set(f"{base}{new_ext}")
def browse_source(self):
"""选择源文件"""
source_type, _ = self.get_conversion_info()
file_type = self.file_extensions[source_type]
path = filedialog.askopenfilename(
title="选择源文件",
filetypes=[file_type, ("所有文件", "*.*")]
)
if path:
self.source_path.set(path)
self.auto_suggest_target_path(path)
def auto_suggest_target_path(self, source_path):
"""自动生成目标路径建议"""
source_type, target_type = self.get_conversion_info()
base = os.path.splitext(source_path)[0]
new_ext = self.file_extensions[target_type][1]
suggested_path = f"{base}_converted{new_ext}"
self.target_path.set(suggested_path)
def browse_target(self):
"""选择目标路径"""
_, target_type = self.get_conversion_info()
file_type = self.file_extensions[target_type]
path = filedialog.asksaveasfilename(
title="保存目标文件",
defaultextension=file_type[1],
filetypes=[file_type, ("所有文件", "*.*")]
)
if path:
self.target_path.set(path)
def clear_preview(self):
"""清空预览"""
self.ptable.clearTable()
self.ptable.model.df = pd.DataFrame()
self.ptable.redraw()
def load_preview_data(self):
"""加载预览数据"""
source_path = self.source_path.get()
if not source_path:
return None
try:
source_type, _ = self.get_conversion_info()
if source_type == "excel":
return pd.read_excel(source_path, nrows=10)
elif source_type == "csv":
return pd.read_csv(source_path, nrows=10)
elif source_type == "json":
return pd.read_json(source_path).head(10)
elif source_type == "sqlite":
conn = sqlite3.connect(source_path)
table_name = self.table_name.get() or self.detect_table_name(conn)
if table_name:
return pd.read_sql_query(f"SELECT * FROM {table_name} LIMIT 10", conn)
return None
except Exception as e:
messagebox.showerror("预览错误", f"无法加载数据: {str(e)}")
return None
def detect_table_name(self, conn):
"""自动检测数据库表名"""
cursor = conn.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
tables = cursor.fetchall()
if tables:
return askstring("选择表", "检测到多个表，请选择:", initialvalue=tables[0][0])
return None
def update_preview(self, *args):
"""更新数据预览"""
df = self.load_preview_data()
if df is not None:
self.ptable.model.df = df
self.ptable.redraw()
self.ptable.autoResizeColumns()
def start_conversion(self):
"""执行转换操作"""
source_path = self.source_path.get()
target_path = self.target_path.get()
conversion_type = self.conversion_type.get()
table_name = self.table_name.get()
try:
# 自动修正目标路径后缀
_, target_type = self.get_conversion_info()
target_ext = self.file_extensions[target_type][1]
if not target_path.endswith(target_ext):
target_path = f"{os.path.splitext(target_path)[0]}{target_ext}"
self.target_path.set(target_path)
# 执行转换逻辑
if conversion_type == "excel2json":
df = pd.read_excel(source_path)
df.to_json(target_path, orient='records', indent=4)
elif conversion_type == "json2excel":
df = pd.read_json(source_path)
df.to_excel(target_path, index=False)
elif conversion_type == "csv2json":
df = pd.read_csv(source_path)
df.to_json(target_path, orient='records', indent=4)
elif conversion_type == "json2csv":
df = pd.read_json(source_path)
df.to_csv(target_path, index=False)
elif conversion_type == "sqlite2excel":
conn = sqlite3.connect(source_path)
table_name = table_name or self.detect_table_name(conn)
df = pd.read_sql_query(f"SELECT * FROM {table_name}", conn)
df.to_excel(target_path, index=False)
elif conversion_type == "excel2sqlite":
df = pd.read_excel(source_path)
conn = sqlite3.connect(target_path)
df.to_sql(table_name or "Sheet1", conn, if_exists='replace', index=False)
elif conversion_type == "sqlite2csv":
conn = sqlite3.connect(source_path)
table_name = table_name or self.detect_table_name(conn)
df = pd.read_sql_query(f"SELECT * FROM {table_name}", conn)
df.to_csv(target_path, index=False)
elif conversion_type == "csv2sqlite":
df = pd.read_csv(source_path)
conn = sqlite3.connect(target_path)
df.to_sql(table_name or "CSV_Data", conn, if_exists='replace', index=False)
messagebox.showinfo("成功", f"文件已成功保存到:\n{target_path}")
self.update_preview()
except Exception as e:
messagebox.showerror("错误", f"转换失败: {str(e)}")
if __name__ == "__main__":
root = tk.Tk()
app = DataConverterApp(root)
root.mainloop()