<think> 我们参考用户的需求和之前的回答，设计一个图形界面工具，用于执行数据清洗和导入数据库的脚本，并弹窗提示结果。 我们将使用Tkinter创建界面，因为它简单且内置于Python标准库中。 设计思路： 1. 创建主窗口，包含三个按钮：执行数据清洗、执行数据导入、执行完整流程（清洗+导入）。 2. 每个按钮点击后调用相应的函数，函数中通过subprocess运行对应的脚本（假设脚本路径已知）。 3. 运行脚本后，捕获执行结果（包括成功或失败），并通过messagebox弹窗显示结果。 4. 考虑脚本执行可能产生的输出信息，我们截取部分信息显示在弹窗中（避免过长）。 根据引用[1]和[4]，我们使用tkinter.messagebox来显示弹窗；根据引用[3]，我们可以使用PySimpleGUI，但用户要求用弹窗提示，且之前代码已用tkinter，故继续使用tkinter。 我们将按照以下步骤编写代码： 步骤1：导入必要的库 步骤2：创建GUI界面，包含三个按钮 步骤3：定义执行脚本的函数，该函数使用subprocess运行脚本，并返回执行状态和输出 步骤4：为每个按钮绑定事件处理函数，在事件处理函数中调用执行脚本的函数，并根据返回结果弹出提示 注意：我们假设数据清洗脚本和导入脚本已经存在，并且可以在当前环境中运行（即依赖已安装） 代码结构如下： 我们将创建一个类DataProcessorApp，包含以下方法： - __init__: 初始化界面 - run_script(script_path): 执行指定脚本，返回成功状态和输出 - show_result(success, message, process_name): 根据执行结果显示弹窗 - run_data_cleaning(): 执行数据清洗脚本 - run_data_import(): 执行数据导入脚本 - run_full_process(): 执行清洗然后导入（如果清洗成功） 下面提供完整代码： </think> # Python数据清洗与导入工具实现指南 我将基于您的要求，设计一个图形化工具来管理数据清洗和导入流程。这个工具允许您选择并执行数据清洗脚本，将结果导入数据库，并实时显示运行状态和结果。 ## 解决方案设计 ```mermaid graph TD A[主界面] --> B[脚本选择] B --> C[数据清洗] C --> D{清洗成功?} D -->|是| E[导入数据库] D -->|否| F[显示错误] E --> G{导入成功?} G -->|是| H[显示成功] G -->|否| I[显示错误] H --> J[生成报告] I --> J ``` ## 完整代码实现 ```python import tkinter as tk from tkinter import ttk, messagebox, scrolledtext import subprocess import os import threading import time class DataProcessingApp: def __init__(self, root): self.root = root self.root.title("数据清洗与导入工具") self.root.geometry("800x600") # 脚本配置 self.cleaning_scripts = self.find_scripts("清洗脚本", "clean_*.py") self.import_scripts = self.find_scripts("导入脚本", "import_*.py") # 创建界面 self.create_widgets() # 状态跟踪 self.is_running = False self.current_process = None def find_scripts(self, folder, pattern): """查找指定文件夹中的脚本文件""" if not os.path.exists(folder): os.makedirs(folder) return [f for f in os.listdir(folder) if f.endswith('.py')] def create_widgets(self): """创建界面组件""" # 顶部标题 header = ttk.Label(self.root, text="数据清洗与导入工具", font=("Arial", 16, "bold")) header.pack(pady=20) # 创建主框架 main_frame = ttk.Frame(self.root) main_frame.pack(fill=tk.BOTH, expand=True, padx=20, pady=10) # 左面板 - 脚本选择 left_frame = ttk.LabelFrame(main_frame, text="脚本选择") left_frame.pack(side=tk.LEFT, fill=tk.BOTH, expand=True, padx=10, pady=5) # 清洗脚本选择 ttk.Label(left_frame, text="选择数据清洗脚本:").pack(anchor="w", padx=10, pady=(10, 5)) self.clean_combo = ttk.Combobox(left_frame, values=self.cleaning_scripts) self.clean_combo.pack(fill=tk.X, padx=10, pady=5) # 导入脚本选择 ttk.Label(left_frame, text="选择数据导入脚本:").pack(anchor="w", padx=10, pady=(10, 5)) self.import_combo = ttk.Combobox(left_frame, values=self.import_scripts) self.import_combo.pack(fill=tk.X, padx=10, pady=5) # 按钮区域 btn_frame = ttk.Frame(left_frame) btn_frame.pack(fill=tk.X, padx=10, pady=20) self.clean_btn = ttk.Button(btn_frame, text="执行清洗", command=self.run_cleaning) self.clean_btn.pack(side=tk.LEFT, padx=5) self.import_btn = ttk.Button(btn_frame, text="执行导入", command=self.run_import) self.import_btn.pack(side=tk.LEFT, padx=5) self.full_btn = ttk.Button(btn_frame, text="完整流程", command=self.run_full_process) self.full_btn.pack(side=tk.LEFT, padx=5) self.stop_btn = ttk.Button(btn_frame, text="停止", command=self.stop_process, state=tk.DISABLED) self.stop_btn.pack(side=tk.LEFT, padx=5) # 右面板 - 日志输出 right_frame = ttk.LabelFrame(main_frame, text="执行日志") right_frame.pack(side=tk.RIGHT, fill=tk.BOTH, expand=True, padx=10, pady=5) self.log_text = scrolledtext.ScrolledText(right_frame, wrap=tk.WORD, height=20) self.log_text.pack(fill=tk.BOTH, expand=True, padx=10, pady=10) self.log_text.config(state=tk.DISABLED) # 状态栏 self.status_var = tk.StringVar(value="就绪") status_bar = ttk.Label(self.root, textvariable=self.status_var, relief=tk.SUNKEN, anchor=tk.W) status_bar.pack(side=tk.BOTTOM, fill=tk.X) # 进度条 self.progress = ttk.Progressbar(self.root, orient=tk.HORIZONTAL, length=400, mode='determinate') self.progress.pack(side=tk.BOTTOM, fill=tk.X, padx=20, pady=5) def log_message(self, message): """记录日志消息""" self.log_text.config(state=tk.NORMAL) timestamp = time.strftime("%Y-%m-%d %H:%M:%S") self.log_text.insert(tk.END, f"[{timestamp}] {message}\n") self.log_text.config(state=tk.DISABLED) self.log_text.see(tk.END) # 自动滚动到底部 def update_status(self, message): """更新状态栏""" self.status_var.set(message) self.log_message(message) def run_script(self, script_type, script_name): """执行指定脚本""" if not script_name: messagebox.showwarning("警告", f"请先选择{script_type}脚本") return False, "未选择脚本" folder = "清洗脚本" if script_type == "清洗" else "导入脚本" script_path = os.path.join(folder, script_name) if not os.path.exists(script_path): messagebox.showerror("错误", f"脚本文件不存在: {script_path}") return False, "脚本文件不存在" try: self.update_status(f"开始执行{script_type}脚本: {script_name}") self.progress['value'] = 0 # 在新线程中执行脚本 self.is_running = True self.stop_btn.config(state=tk.NORMAL) def run(): try: process = subprocess.Popen( ["python", script_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1, universal_newlines=True ) self.current_process = process # 实时读取输出 while True: output = process.stdout.readline() if output == '' and process.poll() is not None: break if output: self.log_message(output.strip()) # 检查执行结果 return_code = process.poll() if return_code == 0: self.update_status(f"{script_type}脚本执行成功") self.progress['value'] = 100 return True, "执行成功" else: error_output = process.stderr.read() self.log_message(f"错误: {error_output}") self.update_status(f"{script_type}脚本执行失败") return False, error_output except Exception as e: self.log_message(f"执行异常: {str(e)}") return False, str(e) finally: self.is_running = False self.stop_btn.config(state=tk.DISABLED) self.current_process = None # 启动执行线程 thread = threading.Thread(target=run) thread.daemon = True thread.start() # 模拟进度更新（实际项目中应基于实际进度更新） def update_progress(): if self.is_running: current = self.progress['value'] if current < 90: self.progress['value'] = current + 5 self.root.after(1000, update_progress) self.root.after(1000, update_progress) return True, "执行中" except Exception as e: self.update_status(f"执行失败: {str(e)}") return False, str(e) def stop_process(self): """停止当前执行过程""" if self.current_process: self.current_process.terminate() self.update_status("操作已停止") self.is_running = False self.stop_btn.config(state=tk.DISABLED) def run_cleaning(self): """执行数据清洗脚本""" script_name = self.clean_combo.get() success, message = self.run_script("清洗", script_name) if not success and message != "执行中": messagebox.showerror("清洗失败", f"数据清洗失败:\n{message}") elif success: messagebox.showinfo("清洗成功", "数据清洗完成!") def run_import(self): """执行数据导入脚本""" script_name = self.import_combo.get() success, message = self.run_script("导入", script_name) if not success and message != "执行中": messagebox.showerror("导入失败", f"数据导入失败:\n{message}") elif success: messagebox.showinfo("导入成功", "数据已成功导入数据库!") def run_full_process(self): """执行完整流程：清洗 + 导入""" clean_script = self.clean_combo.get() import_script = self.import_combo.get() if not clean_script or not import_script: messagebox.showwarning("警告", "请先选择清洗和导入脚本") return # 执行清洗 self.update_status("开始完整流程: 数据清洗") success, _ = self.run_script("清洗", clean_script) if not success: messagebox.showerror("流程中断", "数据清洗失败，停止后续操作") return # 执行导入 self.update_status("开始完整流程: 数据导入") success, _ = self.run_script("导入", import_script) if success: messagebox.showinfo("流程完成", "完整流程执行成功!") else: messagebox.showerror("流程中断", "数据导入失败") if __name__ == "__main__": root = tk.Tk() app = DataProcessingApp(root) root.mainloop() ``` ## 使用说明 ### 1. 准备工作 创建两个文件夹存放脚本： ```bash mkdir 清洗脚本 mkdir 导入脚本 ``` ### 2. 创建示例脚本 **清洗脚本示例 (清洗脚本/clean_example.py):** ```python import pandas as pd import time import random print("开始数据清洗...") # 模拟数据处理 for i in range(1, 11): print(f"处理数据块 {i}/10") time.sleep(0.5) # 模拟处理时间 # 模拟数据清洗结果 records = random.randint(1000, 5000) print(f"清洗完成! 共处理 {records} 条记录") ``` **导入脚本示例 (导入脚本/import_example.py):** ```python import time import random print("开始导入数据库...") # 模拟数据库导入 total = random.randint(500, 1500) for i in range(1, 11): processed = int(total * i / 10) print(f"已导入 {processed}/{total} 条记录") time.sleep(0.3) # 模拟导入时间 print("导入完成!") ``` ### 3. 运行工具 ```bash python data_processor.py ``` ### 4. 工具功能说明 1. **脚本选择**： - 从下拉菜单中选择清洗和导入脚本 - 支持添加多个脚本文件（命名格式为clean_*.py和import_*.py） 2. **执行控制**： - 单独执行清洗或导入 - 一键执行完整流程 - 支持停止正在执行的任务 3. **状态监控**： - 实时日志输出 - 进度条显示 - 状态栏提示当前状态 4. **结果反馈**： - 成功时显示成功弹窗 - 失败时显示错误详情 - 完整流程执行结束显示汇总结果 ## 扩展功能建议 1. **脚本参数配置**：添加界面让用户配置脚本运行参数 2. **历史记录**：保存每次执行的日志和结果 3. **电子邮件通知**：执行完成后发送邮件通知 4. **任务调度**：添加定时执行功能 5. **数据库配置管理**：存储和管理数据库连接信息 这个工具提供了灵活的数据处理流程管理，通过图形界面简化了脚本调用过程，实时反馈执行状态，非常适合需要定期执行数据清洗和导入的场景。