#!/usr/bin/env python # -*- coding: utf-8 -*- """ 配置管理工具 用于启动时动态配置数据库连接和API地址 """ import os import sys import json import tkinter as tk from tkinter import ttk, messagebox, filedialog from typing import Dict, Any import configparser class ConfigManager: """配置管理器""" def __init__(self): self.config_file = "config.ini" self.config = configparser.ConfigParser() self.config_data = {} self.load_config() def load_config(self): """加载配置""" if os.path.exists(self.config_file): self.config.read(self.config_file, encoding='utf-8') # 转换为字典格式 for section in self.config.sections(): self.config_data[section] = dict(self.config[section]) else: # 默认配置 self.config_data = { 'database': { 'username': 'root', 'password': 'root', 'host': 'localhost' }, 'api': { 'tf_api_base_url': 'http://111.231.146.230:4080/jeecg-boot' } } def save_config(self): """保存配置""" # 清空现有配置 for section in self.config.sections(): self.config.remove_section(section) # 添加新配置 for section_name, section_data in self.config_data.items(): self.config.add_section(section_name) for key, value in section_data.items(): self.config.set(section_name, key, str(value)) with open(self.config_file, 'w', encoding='utf-8') as f: self.config.write(f) def get_config(self, section: str, key: str, default=None): """获取配置值""" return self.config_data.get(section, {}).get(key, default) def set_config(self, section: str, key: str, value: str): """设置配置值""" if section not in self.config_data: self.config_data[section] = {} self.config_data[section][key] = value class ConfigGUI: """配置GUI界面""" def __init__(self): self.config_manager = ConfigManager() self.root = tk.Tk() self.setup_gui() def setup_gui(self): """设置GUI界面""" self.root.title("VWED任务系统配置工具") self.root.geometry("600x500") self.root.resizable(True, True) # 设置窗口图标（如果存在） try: self.root.iconbitmap("icon.ico") except: pass # 创建主框架 main_frame = ttk.Frame(self.root, padding="10") main_frame.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S)) # 配置行列权重 self.root.columnconfigure(0, weight=1) self.root.rowconfigure(0, weight=1) main_frame.columnconfigure(1, weight=1) # 标题 title_label = ttk.Label(main_frame, text="VWED任务系统配置", font=("Arial", 16, "bold")) title_label.grid(row=0, column=0, columnspan=3, pady=(0, 20)) # 数据库配置区域 db_frame = ttk.LabelFrame(main_frame, text="数据库配置", padding="10") db_frame.grid(row=1, column=0, columnspan=3, sticky=(tk.W, tk.E), pady=(0, 10)) db_frame.columnconfigure(1, weight=1) # 数据库用户名 ttk.Label(db_frame, text="用户名:").grid(row=0, column=0, sticky=tk.W, padx=(0, 10)) self.db_username = tk.StringVar(value=self.config_manager.get_config('database', 'username', 'root')) ttk.Entry(db_frame, textvariable=self.db_username, width=40).grid(row=0, column=1, sticky=(tk.W, tk.E), pady=2) # 数据库密码 ttk.Label(db_frame, text="密码:").grid(row=1, column=0, sticky=tk.W, padx=(0, 10)) self.db_password = tk.StringVar(value=self.config_manager.get_config('database', 'password', 'root')) ttk.Entry(db_frame, textvariable=self.db_password, show="*", width=40).grid(row=1, column=1, sticky=(tk.W, tk.E), pady=2) # 数据库主机 ttk.Label(db_frame, text="主机地址:").grid(row=2, column=0, sticky=tk.W, padx=(0, 10)) self.db_host = tk.StringVar(value=self.config_manager.get_config('database', 'host', 'localhost')) ttk.Entry(db_frame, textvariable=self.db_host, width=40).grid(row=2, column=1, sticky=(tk.W, tk.E), pady=2) # API配置区域 api_frame = ttk.LabelFrame(main_frame, text="API配置", padding="10") api_frame.grid(row=2, column=0, columnspan=3, sticky=(tk.W, tk.E), pady=(0, 10)) api_frame.columnconfigure(1, weight=1) # TF API基础URL ttk.Label(api_frame, text="TF API地址:").grid(row=0, column=0, sticky=tk.W, padx=(0, 10)) self.tf_api_url = tk.StringVar(value=self.config_manager.get_config('api', 'tf_api_base_url', 'http://111.231.146.230:4080/jeecg-boot')) ttk.Entry(api_frame, textvariable=self.tf_api_url, width=40).grid(row=0, column=1, sticky=(tk.W, tk.E), pady=2) # 按钮区域 button_frame = ttk.Frame(main_frame) button_frame.grid(row=3, column=0, columnspan=3, pady=(20, 0)) # 测试连接按钮 ttk.Button(button_frame, text="测试数据库连接", command=self.test_database_connection).pack(side=tk.LEFT, padx=(0, 10)) # 保存配置按钮 ttk.Button(button_frame, text="保存配置", command=self.save_configuration).pack(side=tk.LEFT, padx=(0, 10)) # 启动服务按钮 ttk.Button(button_frame, text="启动服务", command=self.start_service).pack(side=tk.LEFT, padx=(0, 10)) # 退出按钮 ttk.Button(button_frame, text="退出", command=self.root.quit).pack(side=tk.LEFT) # 状态区域 status_frame = ttk.Frame(main_frame) status_frame.grid(row=4, column=0, columnspan=3, sticky=(tk.W, tk.E), pady=(20, 0)) status_frame.columnconfigure(0, weight=1) # 状态文本 self.status_text = tk.Text(status_frame, height=10, width=70) status_scrollbar = ttk.Scrollbar(status_frame, orient=tk.VERTICAL, command=self.status_text.yview) self.status_text.configure(yscrollcommand=status_scrollbar.set) self.status_text.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S)) status_scrollbar.grid(row=0, column=1, sticky=(tk.N, tk.S)) status_frame.rowconfigure(0, weight=1) self.log_message("配置工具已启动") def log_message(self, message: str): """记录日志消息""" self.status_text.insert(tk.END, f"{message}\n") self.status_text.see(tk.END) self.root.update() def test_database_connection(self): """测试数据库连接""" try: import pymysql self.log_message("正在测试数据库连接...") connection = pymysql.connect( host=self.db_host.get(), user=self.db_username.get(), password=self.db_password.get(), charset='utf8mb4' ) with connection.cursor() as cursor: cursor.execute("SELECT VERSION()") version = cursor.fetchone() connection.close() self.log_message(f"数据库连接成功! MySQL版本: {version[0]}") messagebox.showinfo("成功", "数据库连接测试成功!") except Exception as e: error_msg = f"数据库连接失败: {str(e)}" self.log_message(error_msg) messagebox.showerror("错误", error_msg) def save_configuration(self): """保存配置""" try: # 更新配置管理器 self.config_manager.set_config('database', 'username', self.db_username.get()) self.config_manager.set_config('database', 'password', self.db_password.get()) self.config_manager.set_config('database', 'host', self.db_host.get()) self.config_manager.set_config('api', 'tf_api_base_url', self.tf_api_url.get()) # 保存到文件 self.config_manager.save_config() # 设置环境变量 os.environ['DB_USER'] = self.db_username.get() os.environ['DB_PASSWORD'] = self.db_password.get() os.environ['DB_HOST'] = self.db_host.get() os.environ['TF_API_BASE_URL'] = self.tf_api_url.get() self.log_message("配置已保存") messagebox.showinfo("成功", "配置已保存成功!") except Exception as e: error_msg = f"保存配置失败: {str(e)}" self.log_message(error_msg) messagebox.showerror("错误", error_msg) def start_service(self): """启动服务""" try: # 先保存配置 self.save_configuration() self.log_message("正在启动VWED任务系统服务...") # 检查是否为打包后的可执行文件 if getattr(sys, 'frozen', False): # 在打包后的可执行文件中 app_dir = os.path.dirname(sys.executable) main_exe = os.path.join(app_dir, "vwed_task_main.exe") else: # 在开发环境中 app_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) main_exe = os.path.join(app_dir, "app.py") self.log_message(f"启动文件路径: {main_exe}") if os.path.exists(main_exe): import subprocess if main_exe.endswith('.py'): subprocess.Popen([sys.executable, main_exe]) else: subprocess.Popen([main_exe]) self.log_message("服务启动成功!") messagebox.showinfo("成功", "VWED任务系统服务已启动!\n请打开浏览器访问: http://localhost:8000/docs") # 询问是否关闭配置工具 if messagebox.askyesno("提示", "服务已启动，是否关闭配置工具？"): self.root.quit() else: error_msg = f"找不到主程序文件: {main_exe}" self.log_message(error_msg) messagebox.showerror("错误", error_msg) except Exception as e: error_msg = f"启动服务失败: {str(e)}" self.log_message(error_msg) messagebox.showerror("错误", error_msg) def run(self): """运行GUI""" self.root.mainloop() def main(): """主函数""" try: print("正在启动VWED配置工具...") app = ConfigGUI() app.run() except Exception as e: print(f"配置工具启动失败: {e}") import traceback traceback.print_exc() input("按任意键退出...") if __name__ == "__main__": main()